Joseph Doherty
50bf9ac74a
The Flow Debugger trace arm was recorded when a traced write was ISSUED, not when the DCL acknowledged it. A write the device rejected (or that timed out) left the arm standing for the whole TraceArmWindowSeconds (30 s), so the next unrelated subscription update on that attribute was stamped with the debugger's execution id and spawned a traced, parented run — the debugger drew a "device echo" of a write that never landed. The write's continuation runs off the actor's dispatcher, so it cannot touch _traceArms directly; it Tells self a new DisarmTrace message instead, handled in a Receive<>. The disarm is keyed on the writer's execution id, so a late failure cannot steal an arm a newer traced write has since taken. Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
238 lines
10 KiB
C#
238 lines
10 KiB
C#
using Akka.Actor;
|
|
using Akka.TestKit;
|
|
using Akka.TestKit.Xunit2;
|
|
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.Streaming;
|
|
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.Streaming;
|
|
using ZB.MOM.WW.ScadaBridge.TestSupport;
|
|
using System.Text.Json;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;
|
|
|
|
/// <summary>
|
|
/// Flow Debugger design 2026-09-05 §5.3 — the Instance Actor trace arm: a TRACED write to a
|
|
/// data-sourced attribute stamps the next DCL echo for that attribute with the writer's
|
|
/// <c>SourceExecutionId</c> and <c>Trace</c>; the arm is consumed once and expires unconsumed.
|
|
/// Untraced writes never arm (the production-unchanged pin). A traced static write publishes
|
|
/// the flag directly.
|
|
/// </summary>
|
|
public class InstanceActorTraceArmTests : TestKit, IDisposable
|
|
{
|
|
private const string Tag = "ns=3;s=Recipe.Processed";
|
|
private readonly SiteStorageService _storage;
|
|
private readonly ScriptCompilationService _compilationService;
|
|
private readonly SharedScriptLibrary _sharedScriptLibrary;
|
|
private readonly TestLocalDb _localDb;
|
|
|
|
public InstanceActorTraceArmTests()
|
|
{
|
|
_localDb = TestLocalDb.CreateTemp("instance-tracearm-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);
|
|
}
|
|
|
|
void IDisposable.Dispose()
|
|
{
|
|
Shutdown();
|
|
var path = _localDb.Path;
|
|
_localDb.Dispose();
|
|
TestLocalDb.DeleteFiles(path);
|
|
}
|
|
|
|
private static FlattenedConfiguration Config(string instance) => new()
|
|
{
|
|
InstanceUniqueName = instance,
|
|
Attributes =
|
|
[
|
|
new ResolvedAttribute
|
|
{
|
|
CanonicalName = "Processed", Value = "false", DataType = "Boolean",
|
|
DataSourceReference = Tag, BoundDataConnectionName = "PLC"
|
|
},
|
|
new ResolvedAttribute { CanonicalName = "Status", Value = "Idle", DataType = "String" },
|
|
]
|
|
};
|
|
|
|
private (IActorRef Actor, TestProbe Dcl, TestProbe Subscriber) Create(string instance, int armWindowSeconds = 30)
|
|
{
|
|
var options = new SiteRuntimeOptions { StreamBufferSize = 100, TraceArmWindowSeconds = armWindowSeconds };
|
|
var streamManager = new SiteStreamManager(options, NullLogger<SiteStreamManager>.Instance);
|
|
streamManager.Initialize(Sys);
|
|
var dcl = CreateTestProbe();
|
|
var actor = ActorOf(Props.Create(() => new InstanceActor(
|
|
instance, JsonSerializer.Serialize(Config(instance)), _storage, _compilationService,
|
|
_sharedScriptLibrary, streamManager, options, NullLogger<InstanceActor>.Instance, dcl.Ref)));
|
|
var subscriber = CreateTestProbe();
|
|
streamManager.Subscribe(instance, subscriber.Ref);
|
|
return (actor, dcl, subscriber);
|
|
}
|
|
|
|
private static WriteTagRequest ExpectWrite(TestProbe dcl)
|
|
=> dcl.FishForMessage<WriteTagRequest>(_ => true, TimeSpan.FromSeconds(5));
|
|
|
|
private void Write(IActorRef actor, TestProbe dcl, string instance, Guid? source, bool trace)
|
|
{
|
|
actor.Tell(new SetStaticAttributeCommand("c", instance, "Processed", "true", DateTimeOffset.UtcNow,
|
|
SourceExecutionId: source, Trace: trace));
|
|
var write = ExpectWrite(dcl);
|
|
Assert.Equal(Tag, write.TagPath);
|
|
dcl.Reply(new WriteTagResponse(write.CorrelationId, true, null, DateTimeOffset.UtcNow));
|
|
Assert.True(ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5)).Success);
|
|
}
|
|
|
|
/// <summary>Same as <see cref="Write"/> but the DCL rejects the write, so the arm must be dropped.</summary>
|
|
private void FailingWrite(IActorRef actor, TestProbe dcl, string instance, Guid? source, bool trace)
|
|
{
|
|
actor.Tell(new SetStaticAttributeCommand("c", instance, "Processed", "true", DateTimeOffset.UtcNow,
|
|
SourceExecutionId: source, Trace: trace));
|
|
var write = ExpectWrite(dcl);
|
|
Assert.Equal(Tag, write.TagPath);
|
|
dcl.Reply(new WriteTagResponse(write.CorrelationId, false, "rejected", DateTimeOffset.UtcNow));
|
|
Assert.False(ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5)).Success);
|
|
}
|
|
|
|
private static AttributeValueChanged Echo(IActorRef actor, TestProbe subscriber, object? value)
|
|
{
|
|
actor.Tell(new TagValueUpdate("PLC", Tag, value, QualityCode.Good, DateTimeOffset.UtcNow));
|
|
return subscriber.FishForMessage<AttributeValueChanged>(
|
|
m => m.AttributeName == "Processed" && Equals(m.Value, value), TimeSpan.FromSeconds(10));
|
|
}
|
|
|
|
[Fact]
|
|
public void TracedDataWrite_ThenEcho_PublishesChangeStampedWithSourceAndTrace()
|
|
{
|
|
var (actor, dcl, sub) = Create("P1");
|
|
var writer = Guid.NewGuid();
|
|
|
|
Write(actor, dcl, "P1", writer, trace: true);
|
|
// The device clamps/echoes whatever it wants — the stamp does not depend on the value.
|
|
var published = Echo(actor, sub, "clamped");
|
|
|
|
Assert.Equal(writer, published.SourceExecutionId);
|
|
Assert.True(published.Trace);
|
|
}
|
|
|
|
[Fact]
|
|
public void UntracedDataWrite_ThenEcho_PublishesUnstampedChange()
|
|
{
|
|
var (actor, dcl, sub) = Create("P2");
|
|
|
|
// Stamped writer but NOT traced — the ordinary script-write path.
|
|
Write(actor, dcl, "P2", Guid.NewGuid(), trace: false);
|
|
var published = Echo(actor, sub, true);
|
|
|
|
Assert.Null(published.SourceExecutionId);
|
|
Assert.False(published.Trace);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Arch-review F6: the arm is recorded when the write is ISSUED, not when the DCL acknowledges
|
|
/// it. A write the device rejects must drop the arm, otherwise the next unrelated subscription
|
|
/// update on that attribute is stamped with the debugger's execution id for the rest of the
|
|
/// 30 s window — the Flow Debugger would draw a "device echo" of a write that never landed.
|
|
/// </summary>
|
|
[Fact]
|
|
public void TracedDataWrite_ThatFails_DisarmsTheAttribute_SoTheNextEchoIsUnstamped()
|
|
{
|
|
var (actor, dcl, sub) = Create("P6");
|
|
|
|
FailingWrite(actor, dcl, "P6", Guid.NewGuid(), trace: true);
|
|
var published = Echo(actor, sub, "unrelated");
|
|
|
|
Assert.Null(published.SourceExecutionId);
|
|
Assert.False(published.Trace);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A failed write must only drop ITS OWN arm — a newer traced write that succeeded in the
|
|
/// meantime owns the arm and keeps it (the disarm is keyed on the writer's execution id).
|
|
/// </summary>
|
|
[Fact]
|
|
public void FailedWrite_DoesNotDisarm_AnArmTakenByALaterWriter()
|
|
{
|
|
var (actor, dcl, sub) = Create("P7");
|
|
var loser = Guid.NewGuid();
|
|
var winner = Guid.NewGuid();
|
|
|
|
// Issue the traced write that will fail, but hold its DCL reply back. The reply goes to the
|
|
// Ask's temporary ref, so capture it now — a later Reply() would answer the second write.
|
|
actor.Tell(new SetStaticAttributeCommand("c1", "P7", "Processed", "true", DateTimeOffset.UtcNow,
|
|
SourceExecutionId: loser, Trace: true));
|
|
var first = ExpectWrite(dcl);
|
|
var firstAsker = dcl.LastSender;
|
|
|
|
// A second traced write lands and succeeds — it now owns the arm.
|
|
Write(actor, dcl, "P7", winner, trace: true);
|
|
|
|
// Only now does the first write fail; its disarm must not steal the winner's arm.
|
|
firstAsker.Tell(new WriteTagResponse(first.CorrelationId, false, "rejected", DateTimeOffset.UtcNow), dcl.Ref);
|
|
Assert.False(ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5)).Success);
|
|
|
|
var published = Echo(actor, sub, "echo");
|
|
Assert.Equal(winner, published.SourceExecutionId);
|
|
Assert.True(published.Trace);
|
|
}
|
|
|
|
[Fact]
|
|
public void Arm_IsConsumedOnce()
|
|
{
|
|
var (actor, dcl, sub) = Create("P3");
|
|
var writer = Guid.NewGuid();
|
|
|
|
Write(actor, dcl, "P3", writer, trace: true);
|
|
var first = Echo(actor, sub, "one");
|
|
var second = Echo(actor, sub, "two");
|
|
|
|
Assert.Equal(writer, first.SourceExecutionId);
|
|
Assert.True(first.Trace);
|
|
Assert.Null(second.SourceExecutionId);
|
|
Assert.False(second.Trace);
|
|
}
|
|
|
|
[Fact]
|
|
[Trait("Category", "Slow")]
|
|
public void Arm_Expired_IsIgnored()
|
|
{
|
|
var (actor, dcl, sub) = Create("P4", armWindowSeconds: 1);
|
|
|
|
Write(actor, dcl, "P4", Guid.NewGuid(), trace: true);
|
|
Thread.Sleep(1100);
|
|
var published = Echo(actor, sub, true);
|
|
|
|
Assert.Null(published.SourceExecutionId);
|
|
Assert.False(published.Trace);
|
|
}
|
|
|
|
[Fact]
|
|
public void StaticWrite_Traced_PublishesTraceFlag()
|
|
{
|
|
var (actor, _, sub) = Create("P5");
|
|
var writer = Guid.NewGuid();
|
|
|
|
actor.Tell(new SetStaticAttributeCommand("c", "P5", "Status", "Running", DateTimeOffset.UtcNow,
|
|
SourceExecutionId: writer, Trace: true));
|
|
Assert.True(ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5)).Success);
|
|
|
|
var published = sub.FishForMessage<AttributeValueChanged>(
|
|
m => m.AttributeName == "Status" && Equals(m.Value, "Running"), TimeSpan.FromSeconds(10));
|
|
Assert.Equal(writer, published.SourceExecutionId);
|
|
Assert.True(published.Trace);
|
|
|
|
// And an untraced static write publishes Trace=false (pin).
|
|
actor.Tell(new SetStaticAttributeCommand("c2", "P5", "Status", "Idle2", DateTimeOffset.UtcNow, SourceExecutionId: writer));
|
|
ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));
|
|
var plain = sub.FishForMessage<AttributeValueChanged>(
|
|
m => m.AttributeName == "Status" && Equals(m.Value, "Idle2"), TimeSpan.FromSeconds(10));
|
|
Assert.False(plain.Trace);
|
|
}
|
|
}
|