From af5e062ba4af73af1c679842f96abd75f9b3f589 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 16 Jul 2026 05:01:58 -0400 Subject: [PATCH] fix(driver-calc): rebuild calc tag state on ReinitializeAsync + re-register deps post-apply (review HIGH) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../CalculationDriver.cs | 117 +++++++++++++++--- .../Drivers/DriverHostActor.cs | 32 ++++- .../Drivers/DriverInstanceActor.cs | 16 +++ .../Drivers/CalculationDependencyFlowTests.cs | 85 +++++++++++++ 4 files changed, 231 insertions(+), 19 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs index bc551e7c..b48ad942 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs @@ -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 _tagsByRawPath = new(StringComparer.Ordinal); // rawPath (a dependency) → the calc tags that read it AND are change-triggered. Drives OnDependencyValue. private readonly Dictionary> _changeDependents = new(StringComparer.Ordinal); // Union of every tag's declared dependency refs — the interest set the host registers on the mux. - private readonly IReadOnlyCollection _dependencyRefs; + // Reassigned by RebuildTagTable so a redeploy that adds/removes dependencies re-registers on the mux. + private IReadOnlyCollection _dependencyRefs = Array.Empty(); // 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() + /// + /// (Re)build the authored tag table (), the change-trigger fan-out + /// (), the dependency-ref interest set (), + /// and the per-tag state map () from . Called + /// once at construction AND on every 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. + /// Per-tag 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 when invoked post-construction (the ctor runs single-threaded). + /// + /// The (new) bound options carrying the authored raw calc tags. + 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(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() }, + }; + + /// Re-bind the authored raw calc tags from a new merged DriverConfig JSON (the ApplyDelta + /// payload). The ctor-fixed timeout is kept — RunTimeout 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. + private CalculationDriverOptions ParseOptions(string driverConfigJson) + { + if (!string.IsNullOrWhiteSpace(driverConfigJson)) + { + try + { + var dto = JsonSerializer.Deserialize(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 }; + } + + /// 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 + /// CalculationDriverConfigDto. + private sealed class ReinitConfigDto + { + public List? RawTags { get; init; } + public int? RunTimeoutMs { get; init; } } // ---- IDriver ---- @@ -155,10 +232,22 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba /// 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(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index b7fe7ba0..74d0af62 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -559,6 +559,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Receive(ForwardToMux); Receive(ForwardNativeAlarm); Receive(HandleDiscoveredNodes); + Receive(HandleDeltaApplied); Receive(HandleRestartDriver); Receive(HandleReconnectDriver); Receive(HandleRouteNodeWrite); @@ -588,6 +589,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers Receive(ForwardToMux); Receive(ForwardNativeAlarm); Receive(HandleDiscoveredNodes); + Receive(HandleDeltaApplied); Receive(HandleRestartDriver); Receive(HandleReconnectDriver); Receive(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(_ => { }); + // 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(HandleDeltaApplied); Receive(_ => { /* 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. } /// @@ -1786,6 +1793,21 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _log.Debug("DriverHost {Node}: ApplyDelta queued for {Id}", _localNode, spec.DriverInstanceId); } + /// + /// A driver child finished applying an in-place (its + /// 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 + /// . A Tell (no blocking); a no-op for a driver with no + /// adapter (not an , or no mux wired on this node). + /// + 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; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 9c711119..943c8b7d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -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); + /// + /// Sent to the parent () AFTER an has fully + /// re-initialised the driver (i.e. completed). The host uses + /// it to re-register this driver's dependency-consumer mux adapter: a driver may rebuild its declared + /// 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. + /// + /// The instance whose adapter should be re-registered. + public sealed record DeltaApplied(string DriverInstanceId); public sealed record WriteAttribute(string TagId, object Value); public sealed record WriteAttributeResult(bool Success, string? Reason); /// @@ -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) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs index a6ed5b17..2363048f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs @@ -31,6 +31,11 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase return new RawTagEntry(rawPath, json, WriteIdempotent: false); } + /// Build the merged DriverConfig JSON the ApplyDelta path re-binds from (mirrors the deploy + /// artifact folding the driver's RawTags into DriverInstanceSpec.DriverConfig). + 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)); } + + /// + /// REDEPLOY proof (review HIGH): a pure DriverConfig change (calc tags added / removed, a script + /// edited) routes to an in-place — 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. + /// + [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(); + 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)); + } }