Joseph Doherty
1551 lines
63 KiB
C#
1551 lines
63 KiB
C#
using Akka.Actor;
|
||
using Akka.TestKit;
|
||
using Akka.TestKit.Xunit2;
|
||
using Microsoft.Extensions.Logging;
|
||
using Microsoft.Extensions.Logging.Abstractions;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
|
||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
|
||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
|
||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.TestSupport;
|
||
using ZB.MOM.WW.ScadaBridge.TestSupport;
|
||
using System.Text.Json;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;
|
||
|
||
/// <summary>
|
||
/// Tests for InstanceActor: attribute loading, static overrides, and persistence.
|
||
/// </summary>
|
||
public class InstanceActorTests : TestKit, IDisposable
|
||
{
|
||
private readonly SiteStorageService _storage;
|
||
private readonly ScriptCompilationService _compilationService;
|
||
private readonly SharedScriptLibrary _sharedScriptLibrary;
|
||
private readonly SiteRuntimeOptions _options;
|
||
private readonly TestLocalDb _localDb;
|
||
|
||
public InstanceActorTests()
|
||
{
|
||
_localDb = TestLocalDb.CreateTemp("instance-actor-test");
|
||
_storage = new SiteStorageService(
|
||
_localDb.Db,
|
||
NullLogger<SiteStorageService>.Instance);
|
||
_storage.InitializeAsync().GetAwaiter().GetResult();
|
||
_compilationService = new ScriptCompilationService(
|
||
NullLogger<ScriptCompilationService>.Instance);
|
||
_sharedScriptLibrary = new SharedScriptLibrary(
|
||
_compilationService, NullLogger<SharedScriptLibrary>.Instance);
|
||
_options = new SiteRuntimeOptions();
|
||
}
|
||
|
||
private IActorRef CreateInstanceActor(string instanceName, FlattenedConfiguration config)
|
||
{
|
||
return ActorOf(Props.Create(() => new InstanceActor(
|
||
instanceName,
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null, // no stream manager in tests
|
||
_options,
|
||
NullLogger<InstanceActor>.Instance)));
|
||
}
|
||
|
||
void IDisposable.Dispose()
|
||
{
|
||
// TestKit teardown first, so no in-flight actor can reach a disposed ILocalDb;
|
||
// then dispose the database before deleting — the master connection anchors the WAL.
|
||
Shutdown();
|
||
var path = _localDb.Path;
|
||
_localDb.Dispose();
|
||
TestLocalDb.DeleteFiles(path);
|
||
}
|
||
|
||
// ── M1.6: site event log `instance_lifecycle` category ──────────────────
|
||
|
||
[Fact]
|
||
public void Stop_ReleasesTagSubscriptionWithOwnerIdentity()
|
||
{
|
||
var (instance, dcl) = CreateInstanceWithDataSourcedAttribute(new SiteRuntimeOptions());
|
||
dcl.ExpectMsg<SubscribeTagsRequest>();
|
||
Sys.Stop(instance);
|
||
var request = dcl.ExpectMsg<UnsubscribeTagsRequest>();
|
||
Assert.Equal("conn-1", request.ConnectionName);
|
||
Assert.Equal(instance, dcl.LastSender);
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_Start_EmitsInstanceLifecycleSiteEvent()
|
||
{
|
||
var siteLog = new FakeSiteEventLogger();
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "LifecyclePump",
|
||
Attributes = [new ResolvedAttribute { CanonicalName = "T", Value = "1", DataType = "Int32" }]
|
||
};
|
||
|
||
ActorOf(Props.Create(() => new InstanceActor(
|
||
"LifecyclePump",
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null,
|
||
_options,
|
||
NullLogger<InstanceActor>.Instance,
|
||
null,
|
||
null,
|
||
new SingleServiceProvider(siteLog))));
|
||
|
||
AwaitAssert(() =>
|
||
{
|
||
var rows = siteLog.OfType("instance_lifecycle");
|
||
Assert.Contains(rows, r =>
|
||
r.Severity == "Info" &&
|
||
r.InstanceId == "LifecyclePump" &&
|
||
r.Source == "InstanceActor:LifecyclePump" &&
|
||
r.Message.Contains("started", StringComparison.OrdinalIgnoreCase));
|
||
}, TimeSpan.FromSeconds(2));
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_Stop_EmitsInstanceLifecycleSiteEvent()
|
||
{
|
||
var siteLog = new FakeSiteEventLogger();
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "StoppedPump",
|
||
Attributes = [new ResolvedAttribute { CanonicalName = "T", Value = "1", DataType = "Int32" }]
|
||
};
|
||
|
||
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
||
"StoppedPump",
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null,
|
||
_options,
|
||
NullLogger<InstanceActor>.Instance,
|
||
null,
|
||
null,
|
||
new SingleServiceProvider(siteLog))));
|
||
|
||
// Let PreStart land its started event, then stop the actor.
|
||
AwaitAssert(() => Assert.NotEmpty(siteLog.OfType("instance_lifecycle")),
|
||
TimeSpan.FromSeconds(2));
|
||
Watch(actor);
|
||
actor.Tell(PoisonPill.Instance);
|
||
ExpectTerminated(actor, TimeSpan.FromSeconds(5));
|
||
|
||
AwaitAssert(() =>
|
||
{
|
||
var rows = siteLog.OfType("instance_lifecycle");
|
||
Assert.Contains(rows, r =>
|
||
r.Severity == "Info" &&
|
||
r.InstanceId == "StoppedPump" &&
|
||
r.Message.Contains("stopped", StringComparison.OrdinalIgnoreCase));
|
||
}, TimeSpan.FromSeconds(2));
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_LoadsAttributesFromConfig()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Temperature", Value = "98.6", DataType = "Double" },
|
||
new ResolvedAttribute { CanonicalName = "Status", Value = "Running", DataType = "String" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump1", config);
|
||
|
||
// Query for an attribute that exists
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-1", "Pump1", "Temperature", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<GetAttributeResponse>();
|
||
Assert.True(response.Found);
|
||
Assert.Equal("98.6", response.Value?.ToString());
|
||
Assert.Equal("corr-1", response.CorrelationId);
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_GetAttribute_NotFound_ReturnsFalse()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump1",
|
||
Attributes = []
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump1", config);
|
||
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-2", "Pump1", "NonExistent", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<GetAttributeResponse>();
|
||
Assert.False(response.Found);
|
||
Assert.Null(response.Value);
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_SetStaticAttribute_UpdatesInMemory()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Temperature", Value = "98.6", DataType = "Double" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump1", config);
|
||
|
||
// A static attribute write replies with SetStaticAttributeResponse.
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-3", "Pump1", "Temperature", "100.0", DateTimeOffset.UtcNow));
|
||
var setResponse = ExpectMsg<SetStaticAttributeResponse>();
|
||
Assert.True(setResponse.Success);
|
||
|
||
// Verify the value changed in memory
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-4", "Pump1", "Temperature", DateTimeOffset.UtcNow));
|
||
|
||
var getResponse = ExpectMsg<GetAttributeResponse>();
|
||
Assert.True(getResponse.Found);
|
||
Assert.Equal("100.0", getResponse.Value?.ToString());
|
||
}
|
||
|
||
[Fact]
|
||
public async Task InstanceActor_SetStaticAttribute_PersistsToSQLite()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "PumpPersist1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Temperature", Value = "98.6", DataType = "Double" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("PumpPersist1", config);
|
||
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-persist", "PumpPersist1", "Temperature", "100.0", DateTimeOffset.UtcNow));
|
||
|
||
// A static attribute write replies with SetStaticAttributeResponse once the
|
||
// in-memory state is updated; then wait for the async SQLite persist.
|
||
ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
await Task.Delay(500);
|
||
|
||
// Verify it persisted to SQLite
|
||
var overrides = await _storage.GetStaticOverridesAsync("PumpPersist1");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("100.0", overrides["Temperature"]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Perf remediation (arch-review WP1.5): a burst of rapid static-attribute writes must
|
||
/// coalesce into ONE batched SQLite transaction on a short single-shot timer — mirroring
|
||
/// NativeAlarmActor's MarkDirtyUpsert/FlushDirtyUpserts (P4) shape — rather than a write
|
||
/// per <see cref="SetStaticAttributeCommand"/>. Pinned two ways: (1) nothing is persisted
|
||
/// before the flush interval elapses, even though every write already replied
|
||
/// synchronously and updated in-memory state; (2) only the LATEST value per attribute
|
||
/// survives, across writes to two different attributes coalesced into the same flush.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_RapidStaticWrites_CoalesceIntoSingleFlush()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "PumpCoalesce1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Counter", Value = "0", DataType = "Int32" },
|
||
new ResolvedAttribute { CanonicalName = "Label", Value = "Idle", DataType = "String" }
|
||
]
|
||
};
|
||
|
||
var flushInterval = TimeSpan.FromSeconds(1);
|
||
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
||
"PumpCoalesce1",
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null,
|
||
_options,
|
||
NullLogger<InstanceActor>.Instance,
|
||
null, // dclManager
|
||
null, // healthCollector
|
||
null, // serviceProvider
|
||
flushInterval)));
|
||
|
||
// Five rapid writes to "Counter" plus one to "Label", all sent back-to-back with no
|
||
// delay between them (well inside the 1s coalescing window).
|
||
for (var i = 1; i <= 5; i++)
|
||
{
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
$"corr-counter-{i}", "PumpCoalesce1", "Counter", i.ToString(), DateTimeOffset.UtcNow));
|
||
}
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-label", "PumpCoalesce1", "Label", "Running", DateTimeOffset.UtcNow));
|
||
|
||
// Every write replies synchronously (in-memory state is authoritative regardless of
|
||
// when the buffered persist lands) — six writes, six responses.
|
||
for (var i = 0; i < 6; i++)
|
||
{
|
||
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Success);
|
||
}
|
||
|
||
// Well before the flush interval elapses, NOTHING has reached SQLite yet — proves the
|
||
// writes did not each fire their own persist (the old per-write fire-and-forget shape
|
||
// would already show a row here).
|
||
await Task.Delay(200);
|
||
var beforeFlush = await _storage.GetStaticOverridesAsync("PumpCoalesce1");
|
||
Assert.Empty(beforeFlush);
|
||
|
||
// After the flush interval elapses, exactly one coalesced write has landed: only the
|
||
// LATEST value per attribute survives.
|
||
await Task.Delay(1500);
|
||
var afterFlush = await _storage.GetStaticOverridesAsync("PumpCoalesce1");
|
||
Assert.Equal(2, afterFlush.Count);
|
||
Assert.Equal("5", afterFlush["Counter"]);
|
||
Assert.Equal("Running", afterFlush["Label"]);
|
||
}
|
||
|
||
[Fact]
|
||
public async Task InstanceActor_LoadsStaticOverridesFromSQLite()
|
||
{
|
||
// Pre-populate overrides in SQLite
|
||
await _storage.SetStaticOverrideAsync("PumpOverride1", "Temperature", "200.0");
|
||
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "PumpOverride1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Temperature", Value = "98.6", DataType = "Double" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("PumpOverride1", config);
|
||
|
||
// Wait for the async override loading to complete (PipeTo)
|
||
await Task.Delay(1000);
|
||
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-5", "PumpOverride1", "Temperature", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<GetAttributeResponse>();
|
||
Assert.True(response.Found);
|
||
// The override value should take precedence over the config default
|
||
Assert.Equal("200.0", response.Value?.ToString());
|
||
}
|
||
|
||
[Fact]
|
||
public async Task StaticOverride_ResetOnRedeployment()
|
||
{
|
||
// Set up an override
|
||
await _storage.SetStaticOverrideAsync("PumpRedeploy", "Temperature", "200.0");
|
||
|
||
// Verify override exists
|
||
var overrides = await _storage.GetStaticOverridesAsync("PumpRedeploy");
|
||
Assert.Single(overrides);
|
||
|
||
// Clear overrides BEFORE the successor exists — the ordering DeploymentManager now
|
||
// guarantees (arch-review 2026-08-19 P1-2). It used to clear inside the deploy's
|
||
// persistence task, i.e. after creating the actor, so this "simulation" pinned an
|
||
// ordering the production path did not have; the end-to-end contract lives in
|
||
// RedeployStaticOverrideResetTests.
|
||
await _storage.ClearStaticOverridesAsync("PumpRedeploy");
|
||
|
||
overrides = await _storage.GetStaticOverridesAsync("PumpRedeploy");
|
||
Assert.Empty(overrides);
|
||
|
||
// Create actor with fresh config -- should NOT have the override
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "PumpRedeploy",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Temperature", Value = "98.6", DataType = "Double" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("PumpRedeploy", config);
|
||
|
||
await Task.Delay(1000);
|
||
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-6", "PumpRedeploy", "Temperature", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<GetAttributeResponse>();
|
||
Assert.Equal("98.6", response.Value?.ToString());
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedAttribute_StartsWithUncertainQuality()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Temperature",
|
||
Value = "0",
|
||
DataType = "Double",
|
||
DataSourceReference = "/Motor/Temperature",
|
||
BoundDataConnectionName = "OpcServer1"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump1", config);
|
||
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-quality-1", "Pump1", "Temperature", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<GetAttributeResponse>();
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Uncertain", response.Quality);
|
||
}
|
||
|
||
/// <summary>
|
||
/// SiteRuntime-019: the disable/enable lifecycle is owned entirely by the
|
||
/// Deployment Manager (it stops / re-creates the Instance Actor itself and
|
||
/// replies to the caller). The Instance Actor must NOT handle
|
||
/// <see cref="DisableInstanceCommand"/> / <see cref="EnableInstanceCommand"/>
|
||
/// — the dead handlers that replied with a misleading "success"
|
||
/// acknowledgement were removed. Sending one to the Instance Actor now goes
|
||
/// unhandled and produces no <see cref="InstanceLifecycleResponse"/>.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DoesNotHandleDisableOrEnableCommands()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump1",
|
||
Attributes = []
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump1", config);
|
||
|
||
actor.Tell(new DisableInstanceCommand("cmd-disable", "Pump1", DateTimeOffset.UtcNow));
|
||
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||
|
||
actor.Tell(new EnableInstanceCommand("cmd-enable", "Pump1", DateTimeOffset.UtcNow));
|
||
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||
}
|
||
|
||
[Fact]
|
||
public void InstanceActor_StaticAttribute_StartsWithGoodQuality()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Label",
|
||
Value = "Main Pump",
|
||
DataType = "String"
|
||
// No DataSourceReference — static attribute
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump1", config);
|
||
|
||
actor.Tell(new GetAttributeRequest(
|
||
"corr-quality-2", "Pump1", "Label", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<GetAttributeResponse>();
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Good", response.Quality);
|
||
}
|
||
|
||
/// <summary>
|
||
/// A single PLC tag can back more than one attribute — e.g. two composed
|
||
/// cooling-tank modules whose members both reference the one simulated
|
||
/// <c>ns=3;s=Tank.Level</c> node. A <see cref="TagValueUpdate"/> must fan out
|
||
/// to every attribute that references that tag path, not just the last one
|
||
/// registered: the tag-path → attribute map previously overwrote on a shared
|
||
/// tag, leaving all but one attribute permanently Uncertain.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_TagUpdate_FansOutToEveryAttributeSharingTheTagPath()
|
||
{
|
||
const string sharedTag = "ns=3;s=Tank.Level";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Motor-1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "CoolingTank.Level", Value = "0", DataType = "Int",
|
||
DataSourceReference = sharedTag, BoundDataConnectionName = "PLC"
|
||
},
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "CoolingTank2.Level", Value = "0", DataType = "Int",
|
||
DataSourceReference = sharedTag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
||
"Motor-1",
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null,
|
||
_options,
|
||
NullLogger<InstanceActor>.Instance,
|
||
dcl.Ref)));
|
||
|
||
// On startup the actor subscribes its data-sourced tags through the DCL.
|
||
dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||
|
||
// One value arrives for the tag that both attributes reference.
|
||
actor.Tell(new TagValueUpdate("PLC", sharedTag, 47, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
// BOTH attributes must reflect it — not just the last-registered one.
|
||
foreach (var attrName in new[] { "CoolingTank.Level", "CoolingTank2.Level" })
|
||
{
|
||
actor.Tell(new GetAttributeRequest("corr-fanout", "Motor-1", attrName, DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Found);
|
||
Assert.Equal("47", response.Value?.ToString());
|
||
Assert.Equal("Good", response.Quality);
|
||
}
|
||
}
|
||
|
||
// ── MV-8: data-sourced List attribute coercion ─────────────────────────
|
||
|
||
private IActorRef CreateInstanceActorWithDcl(string instanceName, FlattenedConfiguration config, TestProbe dcl)
|
||
{
|
||
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
||
instanceName,
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null,
|
||
_options,
|
||
NullLogger<InstanceActor>.Instance,
|
||
dcl.Ref)));
|
||
|
||
// On startup the actor subscribes its data-sourced tags through the DCL.
|
||
dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||
return actor;
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-8: when a data-sourced attribute is declared <c>DataType.List</c>, an
|
||
/// incoming OPC UA array value (a CLR array surfaces from the SDK) must be
|
||
/// coerced into a typed <c>List<int></c> whose elements match the
|
||
/// attribute's ElementDataType. The stored value must be a real list — not a
|
||
/// JSON string — so scripts read a typed collection.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedListAttribute_CoercesArrayToTypedList()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Setpoints";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-List",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Setpoints", Value = null,
|
||
DataType = "List", ElementDataType = "Int32",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-List", config, dcl);
|
||
|
||
// OPC UA delivers an array value (CLR array) for the List-typed tag.
|
||
actor.Tell(new TagValueUpdate("PLC", tag, new[] { 10, 20, 30 }, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-list", "Pump-List", "Setpoints", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Good", response.Quality);
|
||
var list = Assert.IsType<List<int>>(response.Value);
|
||
Assert.Equal(new[] { 10, 20, 30 }, list);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-8: array elements coming in as a different CLR type (here, strings that
|
||
/// are valid integers) must still coerce to the declared element type.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedListAttribute_CoercesElementTypes()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Levels";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-ListCoerce",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Levels", Value = null,
|
||
DataType = "List", ElementDataType = "Double",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-ListCoerce", config, dcl);
|
||
|
||
// Elements arrive as ints/strings but the attribute is List<double>.
|
||
actor.Tell(new TagValueUpdate("PLC", tag, new object[] { 1, "2.5", 3 }, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-coerce", "Pump-ListCoerce", "Levels", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Good", response.Quality);
|
||
var list = Assert.IsType<List<double>>(response.Value);
|
||
Assert.Equal(new[] { 1.0, 2.5, 3.0 }, list);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-8: an element that cannot be coerced to the declared element type must
|
||
/// set the attribute quality to <c>Bad</c> and must NOT crash the actor (it
|
||
/// stays alive and continues to answer queries).
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedListAttribute_ElementMismatch_SetsBadQuality_ActorAlive()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Bad";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-ListBad",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Counts", Value = null,
|
||
DataType = "List", ElementDataType = "Int32",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-ListBad", config, dcl);
|
||
Watch(actor);
|
||
|
||
// "not-a-number" cannot be coerced to int → Bad quality, no crash.
|
||
actor.Tell(new TagValueUpdate("PLC", tag, new object[] { 1, "not-a-number", 3 }, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-bad", "Pump-ListBad", "Counts", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Bad", response.Quality);
|
||
|
||
// The actor must still be alive (no crash / restart) and serving.
|
||
ExpectNoTerminated(actor, TimeSpan.FromMilliseconds(500));
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-8 design contract: a failed coercion keeps the PRIOR value. A good
|
||
/// array is delivered first and stored as a typed list; a subsequent array
|
||
/// with a non-coercible element must NOT overwrite that value — the stored
|
||
/// value stays the prior list while the quality flips to <c>Bad</c>.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedListAttribute_BadCoercion_PreservesPriorValue()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Keep";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-ListKeep",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Counts", Value = null,
|
||
DataType = "List", ElementDataType = "Int32",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-ListKeep", config, dcl);
|
||
|
||
// (1) A good array establishes the prior value.
|
||
actor.Tell(new TagValueUpdate("PLC", tag, new[] { 1, 2, 3 }, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
// (2) A second array with a non-coercible element must not overwrite it.
|
||
actor.Tell(new TagValueUpdate("PLC", tag, new object[] { 4, "not-a-number", 6 }, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
// (3) The stored value is still the prior list; quality is Bad.
|
||
actor.Tell(new GetAttributeRequest("corr-keep", "Pump-ListKeep", "Counts", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Bad", response.Quality);
|
||
var list = Assert.IsType<List<int>>(response.Value);
|
||
Assert.Equal(new[] { 1, 2, 3 }, list);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-8 guard: scalar (non-List) data-sourced attributes keep the existing
|
||
/// pass-through behaviour — a scalar value is stored unchanged.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedScalarAttribute_UnchangedByListPath()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Speed";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-Scalar",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Speed", Value = "0", DataType = "Int32",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-Scalar", config, dcl);
|
||
|
||
actor.Tell(new TagValueUpdate("PLC", tag, 1450, QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-scalar", "Pump-Scalar", "Speed", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Good", response.Quality);
|
||
Assert.Equal(1450, response.Value);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV (C1 fix): a WRITE to a data-sourced <c>DataType.List</c> attribute must
|
||
/// send the DCL a TYPED collection (so OPC UA writes an array node), NOT the
|
||
/// canonical JSON string the script layer produced. The script path encodes
|
||
/// <c>List<int></c> to <c>"[10,20,30]"</c>; HandleSetDataAttribute must
|
||
/// decode that back to a typed <c>List<int></c> before building the
|
||
/// WriteTagRequest. We assert the captured WriteTagRequest.Value is the typed
|
||
/// list {10,20,30} — never the string "[10,20,30]".
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedListWrite_SendsTypedArrayToDcl_NotJsonString()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Setpoints";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-ListWrite",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Setpoints", Value = null,
|
||
DataType = "List", ElementDataType = "Int32",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-ListWrite", config, dcl);
|
||
|
||
// Script-style write: ScopeAccessors (AttributeValueCodec.Encode) has
|
||
// already encoded the script's List<int> to the canonical JSON array string,
|
||
// which is an array of element STRINGS (not raw JSON numbers).
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-write", "Pump-ListWrite", "Setpoints", "[\"10\",\"20\",\"30\"]", DateTimeOffset.UtcNow));
|
||
|
||
// The DCL must receive a WriteTagRequest carrying a TYPED collection.
|
||
var write = dcl.ExpectMsg<WriteTagRequest>(TimeSpan.FromSeconds(5));
|
||
Assert.Equal("PLC", write.ConnectionName);
|
||
Assert.Equal(tag, write.TagPath);
|
||
Assert.IsNotType<string>(write.Value);
|
||
var list = Assert.IsType<List<int>>(write.Value);
|
||
Assert.Equal(new[] { 10, 20, 30 }, list);
|
||
|
||
// Complete the Ask so the actor replies success to the caller.
|
||
dcl.Reply(new WriteTagResponse("corr-write", true, null, DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Success);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV (C1 fix): a malformed value written to a data-sourced List attribute
|
||
/// must be REJECTED before reaching the DCL — Success=false and NO
|
||
/// WriteTagRequest is forwarded (mirrors the static-path malformed rejection).
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_DataSourcedListWrite_Malformed_Rejected_NoDclWrite()
|
||
{
|
||
const string tag = "ns=3;s=Pump.Bad";
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-ListWriteBad",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Setpoints", Value = null,
|
||
DataType = "List", ElementDataType = "Int32",
|
||
DataSourceReference = tag, BoundDataConnectionName = "PLC"
|
||
}
|
||
]
|
||
};
|
||
|
||
var dcl = CreateTestProbe();
|
||
var actor = CreateInstanceActorWithDcl("Pump-ListWriteBad", config, dcl);
|
||
|
||
// Malformed JSON (unterminated array, non-int element) → reject the write.
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-bad-write", "Pump-ListWriteBad", "Setpoints", "[\"a\"", DateTimeOffset.UtcNow));
|
||
|
||
var response = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.False(response.Success);
|
||
Assert.NotNull(response.ErrorMessage);
|
||
|
||
// No write must reach the DCL.
|
||
dcl.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||
}
|
||
|
||
private void ExpectNoTerminated(IActorRef actor, TimeSpan within)
|
||
{
|
||
// The actor is Watch()ed; assert no Terminated arrives in the window.
|
||
// (Liveness is also proven by the preceding successful GetAttributeResponse.)
|
||
ExpectNoMsg(within);
|
||
}
|
||
|
||
// ── MV-7: static (authored) List attribute decode ──────────────────────
|
||
|
||
/// <summary>
|
||
/// MV-7: a STATIC List attribute carries its default as the canonical JSON
|
||
/// array string. On load the actor must decode it to a typed list so a
|
||
/// script reading the attribute receives a real collection, not the raw
|
||
/// JSON string.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_StaticListAttribute_LoadsAsTypedList()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-StaticList",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Labels", Value = "[\"a\",\"b\"]",
|
||
DataType = "List", ElementDataType = "String"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-StaticList", config);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-sl", "Pump-StaticList", "Labels", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Good", response.Quality);
|
||
var list = Assert.IsType<List<string>>(response.Value);
|
||
Assert.Equal(new[] { "a", "b" }, list);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-7: a SetStaticAttribute write on a List attribute decodes the canonical
|
||
/// JSON value into a typed list for in-memory reads, but the PERSISTED form
|
||
/// (SQLite static override) must remain the canonical JSON string — never a
|
||
/// CLR-list .ToString().
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_SetStaticListAttribute_ReadsTypedList_PersistsJsonString()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-SetList",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Labels", Value = "[\"a\",\"b\"]",
|
||
DataType = "List", ElementDataType = "String"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-SetList", config);
|
||
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-set-list", "Pump-SetList", "Labels", "[\"x\",\"y\"]", DateTimeOffset.UtcNow));
|
||
var setResponse = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(setResponse.Success);
|
||
|
||
// In-memory read returns a typed list.
|
||
actor.Tell(new GetAttributeRequest("corr-get-list", "Pump-SetList", "Labels", DateTimeOffset.UtcNow));
|
||
var getResponse = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(getResponse.Found);
|
||
var list = Assert.IsType<List<string>>(getResponse.Value);
|
||
Assert.Equal(new[] { "x", "y" }, list);
|
||
|
||
// The persisted form is the canonical JSON string, NOT a CLR-list .ToString().
|
||
await Task.Delay(500);
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-SetList");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("[\"x\",\"y\"]", overrides["Labels"]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-7: a persisted static override for a List attribute is a canonical JSON
|
||
/// string in SQLite; on load it must be decoded to a typed list, the same as
|
||
/// the config default.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_StaticListOverride_LoadsAsTypedList()
|
||
{
|
||
await _storage.SetStaticOverrideAsync("Pump-OverrideList", "Labels", "[\"p\",\"q\"]");
|
||
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-OverrideList",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Labels", Value = "[\"a\",\"b\"]",
|
||
DataType = "List", ElementDataType = "String"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-OverrideList", config);
|
||
|
||
// Wait for the async override load (PipeTo) to apply.
|
||
await Task.Delay(1000);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-ol", "Pump-OverrideList", "Labels", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
var list = Assert.IsType<List<string>>(response.Value);
|
||
Assert.Equal(new[] { "p", "q" }, list);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-7: a malformed stored List value must NOT crash the actor — it loads
|
||
/// with quality Bad and the actor stays alive and answering.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_StaticListAttribute_Malformed_LoadsBadQuality_ActorAlive()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-BadList",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Labels", Value = "[\"a\"", // truncated JSON
|
||
DataType = "List", ElementDataType = "String"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-BadList", config);
|
||
Watch(actor);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-bl", "Pump-BadList", "Labels", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Bad", response.Quality);
|
||
Assert.Null(response.Value);
|
||
|
||
// The actor must still be alive (no crash / restart during construction).
|
||
ExpectNoTerminated(actor, TimeSpan.FromMilliseconds(500));
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-7 guard: a scalar static attribute is unaffected by the List decode
|
||
/// path — it still returns its raw string value.
|
||
/// </summary>
|
||
[Fact]
|
||
public void InstanceActor_StaticScalarAttribute_UnaffectedByListDecode()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-StaticScalar",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Label", Value = "Main Pump", DataType = "String" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-StaticScalar", config);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-ss", "Pump-StaticScalar", "Label", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Good", response.Quality);
|
||
Assert.Equal("Main Pump", response.Value);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-7 (review fix): a SetStaticAttribute write whose value fails to decode as
|
||
/// a list (e.g. truncated JSON) on a List attribute must be REJECTED — reply
|
||
/// Success=false with a clear error and persist NOTHING. The script path always
|
||
/// pre-encodes valid JSON, but the Inbound API / direct-command path can submit
|
||
/// an arbitrary value, so a malformed value must not silently null the in-memory
|
||
/// value, publish "Good" quality, and durably persist a poison override.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_SetStaticListAttribute_Malformed_Rejected_NotPersisted()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-BadSet",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Labels", Value = "[\"a\",\"b\"]",
|
||
DataType = "List", ElementDataType = "String"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-BadSet", config);
|
||
|
||
// Submit a malformed list value (truncated JSON).
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-bad-set", "Pump-BadSet", "Labels", "[\"a\"", DateTimeOffset.UtcNow));
|
||
var setResponse = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.False(setResponse.Success);
|
||
Assert.False(string.IsNullOrWhiteSpace(setResponse.ErrorMessage));
|
||
|
||
// The poison value must NOT have been persisted as a static override.
|
||
await Task.Delay(500);
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-BadSet");
|
||
Assert.Empty(overrides);
|
||
|
||
// A subsequent read returns the untouched config default — not the poison value.
|
||
actor.Tell(new GetAttributeRequest("corr-bad-get", "Pump-BadSet", "Labels", DateTimeOffset.UtcNow));
|
||
var getResponse = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(getResponse.Found);
|
||
var list = Assert.IsType<List<string>>(getResponse.Value);
|
||
Assert.Equal(new[] { "a", "b" }, list);
|
||
}
|
||
|
||
/// <summary>
|
||
/// MV-7 (review fix): an empty-list value "[]" decodes to a non-null empty list
|
||
/// and must be accepted (NOT mistaken for a malformed value, which also decodes
|
||
/// to null). This pins the boundary between the "clearing/empty" and "poison"
|
||
/// cases that both surface as a null decode result.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_SetStaticListAttribute_EmptyList_Accepted()
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-EmptySet",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Labels", Value = "[\"a\",\"b\"]",
|
||
DataType = "List", ElementDataType = "String"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-EmptySet", config);
|
||
|
||
actor.Tell(new SetStaticAttributeCommand(
|
||
"corr-empty-set", "Pump-EmptySet", "Labels", "[]", DateTimeOffset.UtcNow));
|
||
var setResponse = ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(setResponse.Success);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-empty-get", "Pump-EmptySet", "Labels", DateTimeOffset.UtcNow));
|
||
var getResponse = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(getResponse.Found);
|
||
var list = Assert.IsType<List<string>>(getResponse.Value);
|
||
Assert.Empty(list);
|
||
|
||
// The canonical JSON "[]" is persisted unchanged.
|
||
await Task.Delay(500);
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-EmptySet");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("[]", overrides["Labels"]);
|
||
}
|
||
|
||
// ── NJ-4: old-form List static override normalization on load ────────────
|
||
|
||
/// <summary>
|
||
/// NJ-4: an OLD array-of-strings static override (<c>["10","20"]</c>) for an
|
||
/// Int32 List attribute must be re-persisted in the native form (<c>[10,20]</c>)
|
||
/// when the actor loads it at startup. The in-memory read still returns the
|
||
/// typed list {10,20}; the on-disk value is normalized to native JSON.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_OldFormListOverride_NormalizedToNativeOnLoad()
|
||
{
|
||
await _storage.SetStaticOverrideAsync("Pump-OldForm", "Counts", "[\"10\",\"20\"]");
|
||
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-OldForm",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Counts", Value = "[1,2]",
|
||
DataType = "List", ElementDataType = "Int32"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-OldForm", config);
|
||
|
||
// Wait for the async override load (PipeTo) + fire-and-forget normalization.
|
||
await Task.Delay(1000);
|
||
|
||
// In-memory read returns the typed list, decoded from the old form.
|
||
actor.Tell(new GetAttributeRequest("corr-of", "Pump-OldForm", "Counts", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Found);
|
||
var list = Assert.IsType<List<int>>(response.Value);
|
||
Assert.Equal(new[] { 10, 20 }, list);
|
||
|
||
// The on-disk override has been normalized to the native form.
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-OldForm");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("[10,20]", overrides["Counts"]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// NJ-4: a NATIVE-form static override (<c>[10,20]</c>) is already canonical, so
|
||
/// load-time normalization must be a no-op — the on-disk value is unchanged
|
||
/// (idempotent: native → native is byte-identical, so no re-persist occurs).
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_NativeFormListOverride_NotRePersistedOnLoad()
|
||
{
|
||
await _storage.SetStaticOverrideAsync("Pump-Native", "Counts", "[10,20]");
|
||
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-Native",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Counts", Value = "[1,2]",
|
||
DataType = "List", ElementDataType = "Int32"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-Native", config);
|
||
await Task.Delay(1000);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-nat", "Pump-Native", "Counts", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Found);
|
||
var list = Assert.IsType<List<int>>(response.Value);
|
||
Assert.Equal(new[] { 10, 20 }, list);
|
||
|
||
// The native value is left untouched on disk.
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-Native");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("[10,20]", overrides["Counts"]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// NJ-4: a scalar static override is unaffected by the List normalization path —
|
||
/// its on-disk value is left exactly as stored (no native re-encode).
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_ScalarOverride_NotTouchedByListNormalization()
|
||
{
|
||
await _storage.SetStaticOverrideAsync("Pump-ScalarOf", "Temperature", "200.0");
|
||
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-ScalarOf",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute { CanonicalName = "Temperature", Value = "100.0", DataType = "Double" }
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-ScalarOf", config);
|
||
await Task.Delay(1000);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-sof", "Pump-ScalarOf", "Temperature", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Found);
|
||
Assert.Equal("200.0", response.Value);
|
||
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-ScalarOf");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("200.0", overrides["Temperature"]);
|
||
}
|
||
|
||
/// <summary>
|
||
/// NJ-4: a malformed stored List override (truncated JSON) must NOT crash the
|
||
/// actor and must NOT be re-persisted — it loads with Bad quality (as today),
|
||
/// the actor stays alive, and the poison on-disk value is left unchanged.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task InstanceActor_MalformedListOverride_BadQuality_NotRePersisted()
|
||
{
|
||
await _storage.SetStaticOverrideAsync("Pump-BadOf", "Counts", "[\"a\"");
|
||
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Pump-BadOf",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Counts", Value = "[1,2]",
|
||
DataType = "List", ElementDataType = "Int32"
|
||
}
|
||
]
|
||
};
|
||
|
||
var actor = CreateInstanceActor("Pump-BadOf", config);
|
||
Watch(actor);
|
||
await Task.Delay(1000);
|
||
|
||
actor.Tell(new GetAttributeRequest("corr-bof", "Pump-BadOf", "Counts", DateTimeOffset.UtcNow));
|
||
var response = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
|
||
Assert.True(response.Found);
|
||
Assert.Equal("Bad", response.Quality);
|
||
Assert.Null(response.Value);
|
||
|
||
// The actor must still be alive — no crash from the normalization path.
|
||
ExpectNoTerminated(actor, TimeSpan.FromMilliseconds(500));
|
||
|
||
// The malformed value must NOT have been re-persisted (left exactly as stored).
|
||
var overrides = await _storage.GetStaticOverridesAsync("Pump-BadOf");
|
||
Assert.Single(overrides);
|
||
Assert.Equal("[\"a\"", overrides["Counts"]);
|
||
}
|
||
|
||
// ── S4/UA6: tag-subscription retry (failed AND lost responses) ──
|
||
|
||
/// <summary>
|
||
/// Builds an Instance Actor with a single data-sourced attribute and the given
|
||
/// options (so the retry interval is short), returning it plus the DCL probe.
|
||
/// The initial SubscribeTagsRequest is left in the probe's queue for the caller.
|
||
/// </summary>
|
||
private (IActorRef Instance, TestProbe Dcl) CreateInstanceWithDataSourcedAttribute(SiteRuntimeOptions options)
|
||
{
|
||
var config = new FlattenedConfiguration
|
||
{
|
||
InstanceUniqueName = "Sub-1",
|
||
Attributes =
|
||
[
|
||
new ResolvedAttribute
|
||
{
|
||
CanonicalName = "Motor.Temp", Value = "0", DataType = "Int",
|
||
DataSourceReference = "ns=2;s=Motor.Temp", BoundDataConnectionName = "conn-1"
|
||
}
|
||
]
|
||
};
|
||
var dcl = CreateTestProbe();
|
||
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
||
"Sub-1",
|
||
JsonSerializer.Serialize(config),
|
||
_storage,
|
||
_compilationService,
|
||
_sharedScriptLibrary,
|
||
null,
|
||
options,
|
||
NullLogger<InstanceActor>.Instance,
|
||
dcl.Ref)));
|
||
return (actor, dcl);
|
||
}
|
||
|
||
[Fact]
|
||
public void FailedSubscribeResponse_IsRetried()
|
||
{
|
||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 60_000 };
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||
false, "Unknown connection: conn-1", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
|
||
var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5)); // S4 core
|
||
Assert.Equal(first.ConnectionName, retry.ConnectionName);
|
||
Assert.Equal(first.TagPaths, retry.TagPaths);
|
||
}
|
||
|
||
[Fact]
|
||
public void LostSubscribeResponse_IsRetried_AndSuccessStopsRetrying()
|
||
{
|
||
// Lever 1: a lost reply is re-sent only once the outstanding request is presumed lost (ceiling), by design.
|
||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 };
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
// No reply at all (dead-lettered response) -> must re-send.
|
||
var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||
|
||
instance.Tell(new SubscribeTagsResponse(retry.CorrelationId, retry.InstanceUniqueName,
|
||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(600)); // retry timer cancelled
|
||
}
|
||
|
||
// --- Lever 1 (2026-08-28): in-flight guard ---
|
||
|
||
[Fact]
|
||
public void SlowSubscribe_DoesNotStackDuplicates_WhileInFlight()
|
||
{
|
||
// Retry ticks every 200 ms, but an unanswered request blocks a duplicate for 2 s.
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 200,
|
||
TagSubscribeInFlightCeilingMs = 2000,
|
||
};
|
||
var (_, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
// Pre-Lever-1 this window would carry ~4 more requests (one per 200 ms tick).
|
||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1000));
|
||
}
|
||
|
||
[Fact]
|
||
public void SlowSubscribe_ReissuesAfterInFlightCeiling_AndOnlyOnce()
|
||
{
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 200,
|
||
TagSubscribeInFlightCeilingMs = 600,
|
||
};
|
||
var (_, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // after ceiling
|
||
Assert.NotEqual(first.CorrelationId, second.CorrelationId);
|
||
Assert.Equal(first.TagPaths, second.TagPaths);
|
||
// The new request is itself guarded — no third one inside its ceiling window.
|
||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
|
||
}
|
||
|
||
[Fact]
|
||
public void ReplyToSupersededRequest_IsIgnored_AndLatestStillGoverns()
|
||
{
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 200,
|
||
TagSubscribeInFlightCeilingMs = 300,
|
||
};
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||
|
||
// A late Success for the FIRST id is dropped at the pending-map check (its id was
|
||
// removed at the ceiling); the watchdog guarding the SECOND is untouched.
|
||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // second eventually re-sent
|
||
Assert.NotEqual(second.CorrelationId, third.CorrelationId);
|
||
|
||
// Success for the live id stops everything.
|
||
instance.Tell(new SubscribeTagsResponse(third.CorrelationId, third.InstanceUniqueName,
|
||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800));
|
||
}
|
||
|
||
// --- Lever 1 (2026-08-28): backoff on failure, reset on success ---
|
||
|
||
[Fact]
|
||
public void RepeatedFailures_BackOff_AndSuccessResetsToFloor()
|
||
{
|
||
// Floor 200 ms, max 800 ms. Jitter is ±20 % so bound the assertions, don't pin them.
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 800,
|
||
TagSubscribeInFlightCeilingMs = 60_000, // guard never expires; only replies drive retries
|
||
};
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var req = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
var gaps = new List<TimeSpan>();
|
||
for (var i = 0; i < 3; i++)
|
||
{
|
||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||
false, "gateway busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||
req = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||
gaps.Add(sw.Elapsed);
|
||
}
|
||
|
||
// Expected nominal gaps: 400, 800(cap), 800(cap) — each ±20 % jitter and scheduler slack.
|
||
Assert.InRange(gaps[0].TotalMilliseconds, 300, 560);
|
||
// gaps[1] compounds two jitter draws: 2×(2×200×0.8)×0.8 = 512 ms minimum.
|
||
Assert.InRange(gaps[1].TotalMilliseconds, 500, 1100);
|
||
Assert.InRange(gaps[2].TotalMilliseconds, 600, 1100);
|
||
|
||
// Success resets: no further request follows a confirmed subscription.
|
||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1200));
|
||
}
|
||
|
||
[Fact]
|
||
public void FailureReply_ReissuesEvenThoughGuardHasNotExpired()
|
||
{
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 200,
|
||
TagSubscribeInFlightCeilingMs = 60_000,
|
||
};
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||
false, "Unknown connection: conn-1", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
|
||
// A definite failure clears the in-flight slot, so the watchdog re-sends.
|
||
var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||
Assert.NotEqual(first.CorrelationId, retry.CorrelationId);
|
||
}
|
||
|
||
// --- Lever 1 (2026-08-28): Good quality resets the backoff ---
|
||
|
||
[Fact]
|
||
public void FlappingGood_DoesNotDeferTheRetry()
|
||
{
|
||
// A burst of Good events faster than the floor must not keep re-arming the watchdog
|
||
// from "now" — after the first reset the interval is at the floor and later Goods are
|
||
// no-ops, so the retry still lands within one floor interval of the first Good.
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 300,
|
||
TagSubscribeRetryMaxIntervalMs = 5000,
|
||
TagSubscribeInFlightCeilingMs = 60_000,
|
||
};
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var req = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
for (var i = 0; i < 3; i++)
|
||
{
|
||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
req = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||
}
|
||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
|
||
// Twelve Goods over ~1.2 s — a naive re-arm-from-now on each would push the tick to ≥1.5 s.
|
||
for (var i = 0; i < 12; i++)
|
||
{
|
||
instance.Tell(new ConnectionQualityChanged("conn-1", QualityCode.Good, DateTimeOffset.UtcNow));
|
||
Thread.Sleep(100);
|
||
}
|
||
|
||
// The retry must have landed DURING the burst (~300 ms after the first Good), so the probe
|
||
// already holds it — assert that directly rather than timing the burst's Thread.Sleeps.
|
||
dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromMilliseconds(50));
|
||
}
|
||
|
||
[Fact]
|
||
public void ConnectionBackToGood_ResetsBackoff_AndRetriesPromptly()
|
||
{
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 5000,
|
||
TagSubscribeInFlightCeilingMs = 60_000,
|
||
};
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
// Drive the interval up: 3 failures → 400, 800, 1600 (nominal).
|
||
var req = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
for (var i = 0; i < 3; i++)
|
||
{
|
||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
req = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||
}
|
||
// One more failure would nominally wait ~3.2 s. Instead the connection recovers:
|
||
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
instance.Tell(new ConnectionQualityChanged("conn-1", QualityCode.Good, DateTimeOffset.UtcNow));
|
||
|
||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||
dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(2));
|
||
Assert.True(sw.Elapsed < TimeSpan.FromMilliseconds(1000), $"retry took {sw.Elapsed} — backoff not reset");
|
||
}
|
||
|
||
[Fact]
|
||
public void LateReplyForFailedAndReissuedRequest_IsIgnored()
|
||
{
|
||
var options = new SiteRuntimeOptions
|
||
{
|
||
TagSubscribeRetryIntervalMs = 200,
|
||
TagSubscribeRetryMaxIntervalMs = 200,
|
||
// Short ceiling on purpose: the second request stays in flight (nothing ever
|
||
// answers it), so the in-flight guard is what re-sends. With a 60 s ceiling the
|
||
// guard would hold for 60 s and the test could not tell "late reply ignored"
|
||
// apart from "late reply wrongly cleared the slot" — both look like silence.
|
||
TagSubscribeInFlightCeilingMs = 500,
|
||
};
|
||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||
|
||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||
|
||
// Duplicate/late Success for the FIRST id: already removed from the pending map → ignored,
|
||
// so the watchdog guarding the SECOND is untouched and still re-sends once its
|
||
// in-flight ceiling expires. Had the late reply been honoured it would have cancelled
|
||
// that watchdog and cleared the slot, and no third request would ever be sent.
|
||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||
var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||
Assert.NotEqual(second.CorrelationId, third.CorrelationId);
|
||
}
|
||
}
|