feat(site-runtime): InstanceActor trace arm stamps the device echo of a traced write; Trace rides on published changes
Flow Debugger (design 2026-09-05 §5.3): a traced SetStaticAttributeCommand on a data-sourced attribute records (attribute → writer, expiry) for TraceArmWindowSeconds (new option, default 30, validated > 0); the next DCL echo for that attribute is published with the writer's SourceExecutionId + Trace whatever its value, and the arm is consumed once. Static writes publish Trace directly. Untraced writes never arm (pinned). Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP
This commit is contained in:
@@ -172,6 +172,15 @@ public class InstanceActor : ReceiveActor
|
|||||||
|
|
||||||
// DCL manager actor reference for subscribing to tag values
|
// DCL manager actor reference for subscribing to tag values
|
||||||
private readonly IActorRef? _dclManager;
|
private readonly IActorRef? _dclManager;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Flow Debugger trace arm (design 2026-09-05 §5.3): attribute → (writer's ExecutionId,
|
||||||
|
/// expiry) recorded when a TRACED script write targets a data-sourced attribute. The next
|
||||||
|
/// DCL value change for that attribute is published stamped with the writer's id and
|
||||||
|
/// <c>Trace</c>, and the arm is consumed. Bounded by the attribute count (one entry per
|
||||||
|
/// attribute, newest wins); no sweep — an unconsumed arm is overwritten or expires in place.
|
||||||
|
/// </summary>
|
||||||
|
private readonly Dictionary<string, (Guid SourceExecutionId, DateTimeOffset ExpiresAt)> _traceArms = new();
|
||||||
// Maps each tag path to every attribute canonical name that references it.
|
// Maps each tag path to every attribute canonical name that references it.
|
||||||
// A tag path can back more than one attribute (e.g. two composed modules
|
// A tag path can back more than one attribute (e.g. two composed modules
|
||||||
// whose members reference the same PLC node), so a tag value update must
|
// whose members reference the same PLC node), so a tag value update must
|
||||||
@@ -592,7 +601,10 @@ public class InstanceActor : ReceiveActor
|
|||||||
// execution's id onto the change so a child Alarm Actor this write
|
// execution's id onto the change so a child Alarm Actor this write
|
||||||
// trips can chain its on-trigger script run under the writer. Null for
|
// trips can chain its on-trigger script run under the writer. Null for
|
||||||
// writes with no audited origin (central Test Run, deployment tooling).
|
// writes with no audited origin (central Test Run, deployment tooling).
|
||||||
SourceExecutionId: command.SourceExecutionId);
|
SourceExecutionId: command.SourceExecutionId,
|
||||||
|
// Flow Debugger (design §5.2): a traced write's change carries the flag so the
|
||||||
|
// value-change run it spawns is traced and parented to the writer.
|
||||||
|
Trace: command.Trace);
|
||||||
|
|
||||||
PublishAndNotifyChildren(changed);
|
PublishAndNotifyChildren(changed);
|
||||||
|
|
||||||
@@ -742,6 +754,14 @@ public class InstanceActor : ReceiveActor
|
|||||||
writeValue,
|
writeValue,
|
||||||
DateTimeOffset.UtcNow);
|
DateTimeOffset.UtcNow);
|
||||||
|
|
||||||
|
// Flow Debugger trace arm (design §5.3): only a TRACED write with a known writer arms;
|
||||||
|
// the device echo that follows is stamped in HandleTagValueUpdate. Untraced writes
|
||||||
|
// never touch the map, so production behaviour is unchanged.
|
||||||
|
if (command.Trace && command.SourceExecutionId is { } traceSource)
|
||||||
|
{
|
||||||
|
_traceArms[attributeName] = (traceSource, DateTimeOffset.UtcNow.AddSeconds(_options.TraceArmWindowSeconds));
|
||||||
|
}
|
||||||
|
|
||||||
// Ask the DCL and pipe the result back to the original caller. The DCL
|
// Ask the DCL and pipe the result back to the original caller. The DCL
|
||||||
// returns the failure synchronously so the script can handle it.
|
// returns the failure synchronously so the script can handle it.
|
||||||
_dclManager.Ask<WriteTagResponse>(writeRequest, TimeSpan.FromSeconds(30))
|
_dclManager.Ask<WriteTagResponse>(writeRequest, TimeSpan.FromSeconds(30))
|
||||||
@@ -1035,6 +1055,17 @@ public class InstanceActor : ReceiveActor
|
|||||||
// O(1) lookup off the hot ingest path (was a linear FirstOrDefault).
|
// O(1) lookup off the hot ingest path (was a linear FirstOrDefault).
|
||||||
_resolvedAttributeByName.TryGetValue(attrName, out var resolved);
|
_resolvedAttributeByName.TryGetValue(attrName, out var resolved);
|
||||||
|
|
||||||
|
// Flow Debugger trace arm (design §5.3): consume the arm for this attribute — the
|
||||||
|
// echo is stamped whatever its value (a device that clamps or rejects the write
|
||||||
|
// still shows what it did) — unless the arm has already expired.
|
||||||
|
Guid? traceSource = null;
|
||||||
|
var traced = false;
|
||||||
|
if (_traceArms.Remove(attrName, out var arm) && arm.ExpiresAt > DateTimeOffset.UtcNow)
|
||||||
|
{
|
||||||
|
traceSource = arm.SourceExecutionId;
|
||||||
|
traced = true;
|
||||||
|
}
|
||||||
|
|
||||||
// A List-typed attribute coerces the incoming OPC UA array
|
// A List-typed attribute coerces the incoming OPC UA array
|
||||||
// (a CLR array/IEnumerable from the SDK) into a typed List<T>. On an
|
// (a CLR array/IEnumerable from the SDK) into a typed List<T>. On an
|
||||||
// element-type mismatch we set the attribute's quality to Bad, log a
|
// element-type mismatch we set the attribute's quality to Bad, log a
|
||||||
@@ -1045,7 +1076,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
{
|
{
|
||||||
HandleAttributeValueChanged(new AttributeValueChanged(
|
HandleAttributeValueChanged(new AttributeValueChanged(
|
||||||
_instanceUniqueName, update.TagPath, attrName,
|
_instanceUniqueName, update.TagPath, attrName,
|
||||||
typedList, QualityToString(update.Quality), update.Timestamp));
|
typedList, QualityToString(update.Quality), update.Timestamp,
|
||||||
|
SourceExecutionId: traceSource, Trace: traced));
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
@@ -1062,7 +1094,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
// non-change.
|
// non-change.
|
||||||
PublishAndNotifyChildren(new AttributeValueChanged(
|
PublishAndNotifyChildren(new AttributeValueChanged(
|
||||||
_instanceUniqueName, update.TagPath, attrName,
|
_instanceUniqueName, update.TagPath, attrName,
|
||||||
currentValue, "Bad", update.Timestamp), evaluateWaiters: false);
|
currentValue, "Bad", update.Timestamp,
|
||||||
|
SourceExecutionId: traceSource, Trace: traced), evaluateWaiters: false);
|
||||||
}
|
}
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
@@ -1076,7 +1109,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
|
|
||||||
HandleAttributeValueChanged(new AttributeValueChanged(
|
HandleAttributeValueChanged(new AttributeValueChanged(
|
||||||
_instanceUniqueName, update.TagPath, attrName,
|
_instanceUniqueName, update.TagPath, attrName,
|
||||||
value, QualityToString(update.Quality), update.Timestamp));
|
value, QualityToString(update.Quality), update.Timestamp,
|
||||||
|
SourceExecutionId: traceSource, Trace: traced));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -232,4 +232,15 @@ public class SiteRuntimeOptions
|
|||||||
/// (the row-67 <c>EnsureInstanceStartedCommand</c> path is unaffected). Default: 60.
|
/// (the row-67 <c>EnsureInstanceStartedCommand</c> path is unaffected). Default: 60.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public int InstanceSweepIntervalSeconds { get; set; } = 60;
|
public int InstanceSweepIntervalSeconds { get; set; } = 60;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Flow Debugger trace arm window (design 2026-09-05 §5.3). When a TRACED script write
|
||||||
|
/// targets a data-sourced attribute, the Instance Actor remembers
|
||||||
|
/// <c>(attribute → writer's ExecutionId)</c> for this many seconds; the next DCL value
|
||||||
|
/// change for that attribute — whatever the value — is published stamped with that
|
||||||
|
/// <c>SourceExecutionId</c> and <c>Trace</c>, and the arm is consumed. An arm that expires
|
||||||
|
/// unconsumed emits nothing. Only traced writes arm, so production behaviour is unchanged.
|
||||||
|
/// Default: 30.
|
||||||
|
/// </summary>
|
||||||
|
public int TraceArmWindowSeconds { get; set; } = 30;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -108,5 +108,10 @@ public sealed class SiteRuntimeOptionsValidator : OptionsValidatorBase<SiteRunti
|
|||||||
"instance reconcile-sweep interval and a negative value throws inside the Akka timer. " +
|
"instance reconcile-sweep interval and a negative value throws inside the Akka timer. " +
|
||||||
"0 disables the sweep, which reopens the replication-seeded-config and stale-revision " +
|
"0 disables the sweep, which reopens the replication-seeded-config and stale-revision " +
|
||||||
"self-heals (deferred register rows 78 and 72).");
|
"self-heals (deferred register rows 78 and 72).");
|
||||||
|
|
||||||
|
builder.RequireThat(options.TraceArmWindowSeconds > 0,
|
||||||
|
$"ScadaBridge:SiteRuntime:TraceArmWindowSeconds must be greater than 0 " +
|
||||||
|
$"(was {options.TraceArmWindowSeconds}); it is how long the Instance Actor waits for the " +
|
||||||
|
"device echo of a Flow Debugger traced write before the arm expires (design 2026-09-05 §5.3).");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,178 @@
|
|||||||
|
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);
|
||||||
|
}
|
||||||
|
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
|
||||||
|
[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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -136,6 +136,17 @@ public class SiteRuntimeOptionsValidatorTests
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>A zero cap would shed every trigger, silently disabling all scripts.</summary>
|
/// <summary>A zero cap would shed every trigger, silently disabling all scripts.</summary>
|
||||||
|
[Fact]
|
||||||
|
public void ZeroTraceArmWindowSeconds_IsRejected()
|
||||||
|
{
|
||||||
|
// Flow Debugger design 2026-09-05 §5.3: the arm window must be positive.
|
||||||
|
var result = Validate(new SiteRuntimeOptions { TraceArmWindowSeconds = 0 });
|
||||||
|
|
||||||
|
Assert.True(result.Failed);
|
||||||
|
Assert.Contains("TraceArmWindowSeconds", result.FailureMessage);
|
||||||
|
Assert.Equal(30, new SiteRuntimeOptions().TraceArmWindowSeconds);
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void ZeroMaxConcurrentRunsPerScript_IsRejected()
|
public void ZeroMaxConcurrentRunsPerScript_IsRejected()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user