dohertj2/lmxopcua
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(driver-calc): rebuild calc tag state on ReinitializeAsync + re-register deps post-apply (review HIGH)

Browse Source 
A calc-tag add/remove/edit or a script edit changes the merged DriverConfig,
which DriverSpawnPlanner routes to an in-place ApplyDelta (not a respawn), i.e.
CalculationDriver.ReinitializeAsync. The old ReinitializeAsync ignored the new
config: the tag table + dependency-ref set were built once at construction and
never rebuilt, so after a redeploy added tags never computed, removed tags kept
publishing, and edited scripts served stale results (deploy reported success).

Part 1 (HIGH): extract RebuildTagTable(config) — shared by the ctor and reinit
so they cannot drift — and have ReinitializeAsync re-bind the new config
(ParseOptions) and rebuild _tagsByRawPath/_changeDependents/_dependencyRefs/
_stateByRawPath under _lock, preserving last-known state for surviving tags,
dropping removed tags, and starting new tags fresh. _dependencyRefs now reflects
the new authored tags. Corrected the now-false ReinitializeAsync comment.

Part 2 (MEDIUM ordering hazard): the host's ReRegister was Tell-ed synchronously
alongside the ApplyDelta Tell, so the adapter re-read DependencyRefs before the
async ReinitializeAsync rebuilt them (stale set, no self-correction). Now
DriverInstanceActor sends DeltaApplied(driverInstanceId) to DriverHostActor
AFTER ReinitializeAsync completes, and HandleDeltaApplied re-registers only that
driver's adapter — sequenced after the rebuild. Removed the inline loop.

Test: added a redeploy test to CalculationDependencyFlowTests exercising the
real DependencyMuxActor + adapter + driver — a new tag reading a NEW dep flows +
computes, a removed tag stops publishing, an edited script serves the new
result, and DependencyRefs re-registers the new set. Fails on pre-fix code
(DependencyRefs stays stale ["src/a"]), passes after the fix.

Claude-Session: https://claude.ai/code/session_01LVneM3eh1UtJxEisFXgmox
This commit is contained in:
Joseph Doherty
2026-07-16 05:01:58 -04:00
parent 2e6e89f4a5
commit af5e062ba4
4 changed files with 231 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs
+103 -14
View File
@@ -1,3 +1,5 @@
using System.Text.Json;
using System.Text.Json.Serialization;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Serilog;
@@ -46,15 +48,17 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
private readonly ScriptRootLogger _scriptRoot;
private readonly CalculationEvaluator _evaluator;
// Authored calc tags, keyed by RawPath (identity + evaluation id). Built once at construction so
// DependencyRefs is available BEFORE InitializeAsync (the host reads it when spawning the mux-adapter).
// Authored calc tags, keyed by RawPath (identity + evaluation id). Built at construction so
// DependencyRefs is available BEFORE InitializeAsync (the host reads it when spawning the mux-adapter),
// and REBUILT on every ReinitializeAsync (an in-place ApplyDelta re-binds an edited calc-tag set).
private readonly Dictionary<string, CalculationTagDefinition> _tagsByRawPath = new(StringComparer.Ordinal);
// rawPath (a dependency) → the calc tags that read it AND are change-triggered. Drives OnDependencyValue.
private readonly Dictionary<string, List<string>> _changeDependents = new(StringComparer.Ordinal);
// Union of every tag's declared dependency refs — the interest set the host registers on the mux.
private readonly IReadOnlyCollection<string> _dependencyRefs;
// Reassigned by RebuildTagTable so a redeploy that adds/removes dependencies re-registers on the mux.
private IReadOnlyCollection<string> _dependencyRefs = Array.Empty<string>();
// Live evaluation state. Guarded by _lock: OnDependencyValue (actor thread) and timer ticks
// (pool threads) can both drive an evaluation, so all state mutation is serialized.
@@ -94,16 +98,29 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
_scriptRoot,
_options.RunTimeout);
BuildTagTable();
_dependencyRefs = _tagsByRawPath.Values
.SelectMany(t => t.DependencyRefs)
.Distinct(StringComparer.Ordinal)
.ToArray();
RebuildTagTable(_options);
}
private void BuildTagTable()
/// <summary>
/// (Re)build the authored tag table (<see cref="_tagsByRawPath"/>), the change-trigger fan-out
/// (<see cref="_changeDependents"/>), the dependency-ref interest set (<see cref="_dependencyRefs"/>),
/// and the per-tag state map (<see cref="_stateByRawPath"/>) from <paramref name="config"/>. Called
/// once at construction AND on every <see cref="ReinitializeAsync"/> so an in-place ApplyDelta — a
/// calc-tag add/remove/edit, or a script-source edit — actually takes effect. Extracted so the ctor
/// and reinit paths share ONE construction routine and cannot drift.
/// <para>Per-tag <see cref="TagState"/> is preserved for a tag that survives the edit (cheap last-known
/// value carry-over); a brand-new tag starts fresh; a removed tag's state is dropped. MUST be called
/// under <see cref="_lock"/> when invoked post-construction (the ctor runs single-threaded).</para>
/// </summary>
/// <param name="config">The (new) bound options carrying the authored raw calc tags.</param>
private void RebuildTagTable(CalculationDriverOptions config)
{
foreach (var entry in _options.RawTags)
_tagsByRawPath.Clear();
_changeDependents.Clear();
// Retain surviving tags' state; anything not re-authored is dropped when we swap this in below.
var retainedState = new Dictionary<string, TagState>(StringComparer.Ordinal);
foreach (var entry in config.RawTags)
{
if (!CalculationTagDefinition.TryFromTagConfig(entry.TagConfig, entry.RawPath, out var def))
{
@@ -120,7 +137,9 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
_driverInstanceId, entry.RawPath, def.ScriptId);
}
_tagsByRawPath[entry.RawPath] = def;
_stateByRawPath[entry.RawPath] = new TagState();
// Carry the last-known state forward for a surviving tag; a brand-new tag starts fresh.
retainedState[entry.RawPath] =
_stateByRawPath.TryGetValue(entry.RawPath, out var prior) ? prior : new TagState();
if (def.ChangeTriggered)
{
foreach (var dep in def.DependencyRefs)
@@ -131,6 +150,64 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
}
}
}
// Swap in the retained/new state map — a removed tag's state is dropped so it stops publishing.
_stateByRawPath.Clear();
foreach (var (rawPath, state) in retainedState) _stateByRawPath[rawPath] = state;
_dependencyRefs = _tagsByRawPath.Values
.SelectMany(t => t.DependencyRefs)
.Distinct(StringComparer.Ordinal)
.ToArray();
}
// Reinit config-bind options — mirrors the factory's bind (case-insensitive, enum-string, trailing-comma
// tolerant) so an ApplyDelta payload re-binds identically to the spawn payload.
private static readonly JsonSerializerOptions ReinitJsonOptions = new()
{
PropertyNameCaseInsensitive = true,
ReadCommentHandling = JsonCommentHandling.Skip,
AllowTrailingCommas = true,
Converters = { new JsonStringEnumConverter() },
};
/// <summary>Re-bind the authored raw calc tags from a new merged DriverConfig JSON (the ApplyDelta
/// payload). The ctor-fixed <see cref="CalculationEvaluator"/> timeout is kept — <c>RunTimeout</c> is bound
/// at spawn (a change to it is out of this in-place path's scope). A parse failure logs and keeps the
/// current tags so a malformed delta can never blank the driver.</summary>
private CalculationDriverOptions ParseOptions(string driverConfigJson)
{
if (!string.IsNullOrWhiteSpace(driverConfigJson))
{
try
{
var dto = JsonSerializer.Deserialize<ReinitConfigDto>(driverConfigJson, ReinitJsonOptions);
if (dto is not null)
{
return new CalculationDriverOptions
{
RawTags = dto.RawTags is { Count: > 0 } ? [.. dto.RawTags] : [],
RunTimeout = _options.RunTimeout,
};
}
}
catch (JsonException ex)
{
_logger.LogWarning(ex,
"Calculation {Driver}: reinit config parse failed; keeping the current tag table",
_driverInstanceId);
}
}
return new CalculationDriverOptions { RawTags = _options.RawTags, RunTimeout = _options.RunTimeout };
}
/// <summary>Minimal reinit-bind DTO — the calc driver has no backend config, only the injected raw tags
/// (+ an optional eval-timeout, ignored on reinit since the evaluator is ctor-fixed). Mirrors the factory's
/// <c>CalculationDriverConfigDto</c>.</summary>
private sealed class ReinitConfigDto
{
public List<RawTagEntry>? RawTags { get; init; }
public int? RunTimeoutMs { get; init; }
}
// ---- IDriver ----
@@ -155,10 +232,22 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
/// <inheritdoc />
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// The authored tag table is fixed at construction (a RawTags change respawns the driver, mirroring
// the fleet's factory-built-options pattern). Reinit drops compiled scripts (IScriptCacheOwner
// contract — stops collectible ALCs accreting) + re-arms timers so the calc set stays live.
// A calc-tag add/remove/edit — or a script edit that changes the injected scriptSource — changes the
// merged DriverConfig, which DriverSpawnPlanner routes to an in-place ApplyDelta → THIS call (a
// respawn fires only on a DriverType/ResilienceConfig change). So the authored tag table is NOT
// fixed at construction: we re-bind the new config and REBUILD the tag table + dependency-ref set
// here. Without this, added tags never compute, removed tags keep publishing, and edited scripts
// serve stale results (deploy reports success but nothing changed). After the rebuild _dependencyRefs
// reflects the new authored tags, and the host re-registers the mux interest post-reinit (the
// DriverInstanceActor.DeltaApplied → DriverHostActor path, sequenced AFTER this completes).
StopTimers();
var newOptions = ParseOptions(driverConfigJson);
lock (_lock)
{
RebuildTagTable(newOptions);
}
// Drop compiled scripts (IScriptCacheOwner contract — stops collectible ALCs accreting AND forces a
// recompile from each tag's NEW ScriptSource) + re-arm timers off the rebuilt tag set.
_evaluator.ClearCompiledScripts();
_running = true;
StartTimers();
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+27 -5
View File
@@ -559,6 +559,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
Receive<DriverInstanceActor.AttributeValuePublished>(ForwardToMux);
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
Receive<DriverInstanceActor.DiscoveredNodesReady>(HandleDiscoveredNodes);
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<RestartDriver>(HandleRestartDriver);
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
@@ -588,6 +589,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
Receive<DriverInstanceActor.AttributeValuePublished>(ForwardToMux);
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
Receive<DriverInstanceActor.DiscoveredNodesReady>(HandleDiscoveredNodes);
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<RestartDriver>(HandleRestartDriver);
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
@@ -1262,6 +1264,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
// applied yet, so the equipment can't be resolved). Drop it — the re-discovery loop re-sends it
// and the post-recovery re-apply self-heals it once an apply runs (matches the no-op drops above).
Receive<DriverInstanceActor.DiscoveredNodesReady>(_ => { });
// A late DeltaApplied (an apply completed just before the DB went Stale) — re-register the driver's
// mux adapter anyway; it simply re-reads the driver's current refs (harmless, no DB access).
Receive<DriverInstanceActor.DeltaApplied>(HandleDeltaApplied);
Receive<SubscribeAck>(_ => { /* PubSub ack */ });
Timers.StartPeriodicTimer("retry-db", RetryConfigDbConnection.Instance, ReconnectInterval);
}
@@ -1381,11 +1386,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
foreach (var spec in plan.ToApplyDelta) ApplyChildDelta(spec);
foreach (var spec in plan.ToSpawn) SpawnChild(spec);
// v3 (WP7): re-register every dependency-consumer adapter's interest on each apply — a driver's
// declared dependency refs change when its tags/scripts change. Newly-spawned adapters already
// registered in PreStart; this refreshes the ones an ApplyDelta kept in place. A Tell (no blocking).
foreach (var adapter in _depConsumerAdapters.Values)
adapter.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags.DependencyConsumerMuxAdapter.ReRegister());
// v3 (WP7): a dependency-consumer driver's declared dependency refs change when its tags/scripts are
// edited — the adapter must re-register the NEW set on the mux. Newly-spawned adapters register in
// PreStart; an ApplyDelta'd driver's adapter is re-registered when the child reports it finished
// reinitialising (DriverInstanceActor.DeltaApplied → HandleDeltaApplied). That notification is
// sequenced AFTER ReinitializeAsync rebuilds DependencyRefs — a synchronous re-register HERE would
// re-read the STALE ref set (ApplyDelta runs asynchronously on the child), so it is deliberately NOT
// done inline.
}
/// <summary>
@@ -1786,6 +1793,21 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
_log.Debug("DriverHost {Node}: ApplyDelta queued for {Id}", _localNode, spec.DriverInstanceId);
}
/// <summary>
/// A driver child finished applying an in-place <see cref="DriverInstanceActor.ApplyDelta"/> (its
/// <see cref="IDriver.ReinitializeAsync"/> completed). Re-register THAT driver's dependency-consumer
/// mux adapter so a changed dependency-ref set (calc tags/scripts edited during the redeploy) is
/// adopted on the mux. This runs AFTER reinit — closing the ordering hazard where a synchronous
/// re-register at ApplyDelta dispatch would re-read the driver's stale
/// <see cref="IDependencyConsumer.DependencyRefs"/>. A Tell (no blocking); a no-op for a driver with no
/// adapter (not an <see cref="IDependencyConsumer"/>, or no mux wired on this node).
/// </summary>
private void HandleDeltaApplied(DriverInstanceActor.DeltaApplied msg)
{
if (_depConsumerAdapters.TryGetValue(msg.DriverInstanceId, out var adapter))
adapter.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags.DependencyConsumerMuxAdapter.ReRegister());
}
private void StopChild(string driverInstanceId)
{
if (!_children.TryGetValue(driverInstanceId, out var entry)) return;
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+16
View File
@@ -48,6 +48,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
public sealed record DisconnectObserved(string Reason);
public sealed record ApplyDelta(string DriverConfigJson, CorrelationId Correlation);
public sealed record ApplyResult(bool Success, string? Reason, CorrelationId Correlation);
/// <summary>
/// Sent to the parent (<see cref="DriverHostActor"/>) AFTER an <see cref="ApplyDelta"/> has fully
/// re-initialised the driver (i.e. <see cref="IDriver.ReinitializeAsync"/> completed). The host uses
/// it to re-register this driver's dependency-consumer mux adapter: a driver may rebuild its declared
/// <see cref="IDependencyConsumer.DependencyRefs"/> during reinit (the Calculation driver re-binds its
/// calc tags), so the re-register MUST happen after reinit — NOT synchronously alongside the ApplyDelta
/// dispatch, which would re-read the STALE ref set. Emitted only on a successful apply.
/// </summary>
/// <param name="DriverInstanceId">The instance whose adapter should be re-registered.</param>
public sealed record DeltaApplied(string DriverInstanceId);
public sealed record WriteAttribute(string TagId, object Value);
public sealed record WriteAttributeResult(bool Success, string? Reason);
/// <summary>
@@ -566,6 +576,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
{
await _driver.ReinitializeAsync(msg.DriverConfigJson, CancellationToken.None);
_currentConfigJson = msg.DriverConfigJson;
// The driver may have rebuilt its dependency-consumer interest during reinit (the Calculation
// driver re-binds its calc tags + dependency refs). Notify the parent so it re-registers this
// driver's mux adapter AFTER reinit completed — a synchronous re-register at ApplyDelta-dispatch
// time would re-read the driver's STALE DependencyRefs. Parent Tell (fire-and-forget); the
// continuation resumes on the actor context (no ConfigureAwait) so Context.Parent is live.
Context.Parent.Tell(new DeltaApplied(_driverInstanceId));
replyTo.Tell(new ApplyResult(true, null, msg.Correlation));
}
catch (Exception ex)
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs
+85
View File
@@ -31,6 +31,11 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase
return new RawTagEntry(rawPath, json, WriteIdempotent: false);
}
/// <summary>Build the merged DriverConfig JSON the ApplyDelta path re-binds from (mirrors the deploy
/// artifact folding the driver's RawTags into DriverInstanceSpec.DriverConfig).</summary>
private static string ConfigJson(params RawTagEntry[] tags) =>
System.Text.Json.JsonSerializer.Serialize(new { rawTags = tags });
[Fact]
public async Task Source_value_flows_through_mux_and_adapter_into_the_calc_driver_and_propagates_calc_of_calc()
{
@@ -79,4 +84,84 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase
published.ShouldContain(e => e.FullReference == "calc/engine/b" && Equals(e.Snapshot.Value, 43.0));
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(100));
}
/// <summary>
/// REDEPLOY proof (review HIGH): a pure DriverConfig change (calc tags added / removed, a script
/// edited) routes to an in-place <see cref="CalculationDriver.ReinitializeAsync"/> — NOT a respawn. The
/// driver must rebuild its tag table + dependency-ref set so the new authored set actually takes effect,
/// and the mux adapter must re-register the NEW refs AFTER reinit. This asserts all four failure modes
/// the pre-fix code exhibited: an added tag reading a NEW dep computes; a removed tag stops publishing;
/// an edited script serves the NEW result; and the changed dependency-ref set re-registers on the mux.
/// </summary>
[Fact]
public async Task Redeploy_rebuilds_calc_tag_state_and_reregisters_new_dependency_refs()
{
// Initial authored set: "keep" = src/a*10 (script gets EDITED), "remove" = src/a+5 (gets REMOVED).
var driver = new CalculationDriver(
new CalculationDriverOptions
{
RawTags =
[
CalcTag("calc/keep", "return (double)ctx.GetTag(\"src/a\").Value * 10.0;"),
CalcTag("calc/remove", "return (double)ctx.GetTag(\"src/a\").Value + 5.0;"),
],
},
"calc-drv");
await driver.InitializeAsync("{}", CancellationToken.None);
// Initially the driver only cares about src/a.
driver.DependencyRefs.OrderBy(r => r).ShouldBe(new[] { "src/a" });
var published = new ConcurrentQueue<DataChangeEventArgs>();
var mux = Sys.ActorOf(DependencyMuxActor.Props(), "dep-mux2");
driver.OnDataChange += (_, e) =>
{
published.Enqueue(e);
var quality = e.Snapshot.StatusCode == 0u ? OpcUaQuality.Good : OpcUaQuality.Bad;
mux.Tell(new DriverInstanceActor.AttributeValuePublished("calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts));
};
var adapter = Sys.ActorOf(DependencyConsumerMuxAdapter.Props(driver, mux), "dep-adapter2");
// Leg 1 — before redeploy: src/a=2 → keep=20, remove=7.
AwaitAssert(() =>
{
mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/a", 2.0, OpcUaQuality.Good, Ts));
published.ShouldContain(e => e.FullReference == "calc/keep" && Equals(e.Snapshot.Value, 20.0));
published.ShouldContain(e => e.FullReference == "calc/remove" && Equals(e.Snapshot.Value, 7.0));
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(100));
// ---- REDEPLOY (in-place ApplyDelta) ----
// New authored set: "keep" script EDITED to src/a*1000; "add" is a NEW tag reading a NEW dep src/b;
// "remove" is dropped entirely.
var newConfig = ConfigJson(
CalcTag("calc/keep", "return (double)ctx.GetTag(\"src/a\").Value * 1000.0;"),
CalcTag("calc/add", "return (double)ctx.GetTag(\"src/b\").Value + 100.0;"));
await driver.ReinitializeAsync(newConfig, CancellationToken.None);
// The rebuilt dependency-ref set reflects the NEW authored tags (src/a from keep, src/b from add).
driver.DependencyRefs.OrderBy(r => r).ShouldBe(new[] { "src/a", "src/b" });
// Part 2 sequencing: the host re-registers the adapter AFTER reinit (DeltaApplied → HandleDeltaApplied).
// Mirror that here — re-register so the adapter adopts the NEW ref set (incl. the brand-new src/b).
adapter.Tell(new DependencyConsumerMuxAdapter.ReRegister());
// Leg 2 — after redeploy: the NEW dep src/b flows + the NEW tag computes; the EDITED script serves the
// new result; the REMOVED tag never publishes a value derived from the post-redeploy input.
AwaitAssert(() =>
{
mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/b", 50.0, OpcUaQuality.Good, Ts));
mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/a", 3.0, OpcUaQuality.Good, Ts));
// (a) new dep B flows through the RE-REGISTERED interest → new tag computes: 50 + 100 = 150.
published.ShouldContain(e => e.FullReference == "calc/add" && Equals(e.Snapshot.Value, 150.0));
// (c) edited script produces the NEW result: 3 * 1000 = 3000 (NOT the pre-edit 3 * 10 = 30).
published.ShouldContain(e => e.FullReference == "calc/keep" && Equals(e.Snapshot.Value, 3000.0));
}, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(100));
// (b) the removed tag never recomputes on the post-redeploy src/a=3 (it would be 3 + 5 = 8) and the
// edited "keep" never serves the stale *10 result (3 * 10 = 30). Give any errant compute a moment to land.
await Task.Delay(200);
published.ShouldNotContain(e => e.FullReference == "calc/remove" && Equals(e.Snapshot.Value, 8.0));
published.ShouldNotContain(e => e.FullReference == "calc/keep" && Equals(e.Snapshot.Value, 30.0));
}
}