fix(drivers): mid-connect ApplyDelta adopts config in Calculation, refuses it in Galaxy (#664)

AdoptConfigDuringInit re-runs InitializeAsync(newConfig) and acks success, but
Calculation and Galaxy ignored the JSON there: Calculation kept its old tag
set, and Galaxy connected on its old endpoint. Calculation now rebuilds its tag
table from a config with a body (sharing AdoptConfig with ReinitializeAsync and
stopping any running timer scheduler first). Galaxy, which cannot hot-swap its
options, refuses a non-equivalent one the way ReinitializeAsync does. The actor
now sends DeltaApplied once the adopted init succeeds, so a dependency
consumer's mux interest is re-registered. Unreachable from deploys since #516.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 11:05:58 -04:00
parent 1355049f50
commit 97fcf624ae
8 changed files with 234 additions and 52 deletions
+3
View File
@@ -1168,6 +1168,9 @@ host logger factory (wired in `DriverFactoryBootstrap`; the FOCAS and TwinCAT fa
re-attaches/re-subscribes alarms as well as data, so a failed alarm subscribe is retried every tick.
A Connected in-place `ApplyDelta` (unreachable from deploys since #516) releases both subscriptions at the
driver, reinitialises, then re-subscribes both (#663) — it used to keep handles the reinit had invalidated.
A mid-connect `ApplyDelta` re-runs `InitializeAsync(newConfig)`, so **every driver's `InitializeAsync` must adopt a
config with a body** — Calculation now does, Galaxy (no hot-swap) now refuses a non-equivalent one, and the actor
sends `DeltaApplied` once the adopted init succeeds (#664).
`docs/DriverLifecycle.md`.
- **S7 (#632/#638):** STRING/WSTRING/CHAR/WCHAR and DTL/DATE/TOD/TIME/S5TIME codecs (`S7ValueCodecs`; TOD/TIME/S5TIME
surface as Int32 ms); an undecodable buffer or unrepresentable write is `BadOutOfRange` with no health impact.
+10
View File
@@ -264,6 +264,16 @@ Things to know before touching any of this:
Calculation and Galaxy take their config at construction and ignore the JSON passed to `InitializeAsync`.
Not reachable from a deploy since #516 (a DriverConfig change respawns). Pin:
`An_in_place_ApplyDelta_leaves_one_live_data_and_one_live_alarm_subscription` (one case per wrong fix).
- **A mid-connect `ApplyDelta` re-runs `InitializeAsync(newConfig)` — so every driver's `InitializeAsync` must
honour the JSON (#664).** `AdoptConfigDuringInit` adopts a delta that lands while Connecting/Reconnecting by
re-initialising, and acks on adoption. Every driver re-parses a config with a body (#516/#630) except two that
used to ignore it: **Calculation** now rebuilds its tag table from it (sharing `AdoptConfig` with
`ReinitializeAsync`, and disposing any running timer scheduler first), and **Galaxy**, which cannot hot-swap
its options, now **refuses** a non-equivalent config with the same `NotSupportedException` its
`ReinitializeAsync` throws — it used to connect on the old endpoint while the deploy reported success. A
`"{}"`/empty document still keeps the constructor options. Once the adopted init succeeds the actor sends
`DeltaApplied`, so the host re-registers a dependency consumer's mux interest — the mid-connect path used to
send nothing. A new driver must follow the same rule. Unreachable from a deploy since #516.
- **Restart tests must assert a FUNCTION, never existence.** `RestartDriver_respawns_the_child_without_an_actor_name_collision`
asserts only that a child exists after a restart, and it passed green against the dead driver 03/S16
describes. It is kept — its own subject, the `InvalidActorNameException`, is real — but the load-bearing
@@ -239,6 +239,35 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
return new CalculationDriverOptions { RawTags = _options.RawTags, RunTimeout = _options.RunTimeout };
}
/// <summary>Re-bind the tag table from <paramref name="driverConfigJson"/> — shared by
/// <see cref="ReinitializeAsync"/> and (for a config with a body) <see cref="InitializeAsync"/> so the two
/// adoption paths cannot drift. Timers must already be stopped.</summary>
private void AdoptConfig(string driverConfigJson)
{
var newOptions = ParseOptions(driverConfigJson);
lock (_lock)
{
// Queued evaluations were prepared against the OUTGOING tag table (old script source, old
// dependency set), so they are stale the moment the rebuild lands — drop them rather than
// publish one last pre-edit value after the delta was applied.
_pending.Clear();
RebuildTagTable(newOptions);
}
// Drop compiled scripts (IScriptCacheOwner contract — stops collectible ALCs accreting AND forces a
// recompile from each tag's NEW ScriptSource).
_evaluator.ClearCompiledScripts();
}
/// <summary>True when the supplied DriverConfig JSON carries a real body. The bootstrapper always passes a
/// populated document; unit tests pass <c>"{}"</c> to exercise lifecycle shape — those keep the
/// constructor-supplied tags (an empty document would otherwise bind to an EMPTY tag table).</summary>
private static bool HasConfigBody(string? driverConfigJson)
{
if (string.IsNullOrWhiteSpace(driverConfigJson)) return false;
var trimmed = driverConfigJson.Trim();
return trimmed is not "{}" and not "[]";
}
/// <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>
@@ -257,8 +286,17 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
public string DriverType => DriverTypeNames.Calculation;
/// <inheritdoc />
/// <remarks>
/// #664: <c>InitializeAsync</c> is reachable on a LIVE driver — <c>DriverInstanceActor.AdoptConfigDuringInit</c>
/// re-runs it with a NEW config when an ApplyDelta lands mid-connect — so it adopts a config carrying a
/// real body (the #516 convention every other driver follows), and disposes any running timer scheduler
/// before arming a fresh one (#630: re-initialise without teardown must not leak). A <c>"{}"</c> / empty
/// document keeps the constructor-supplied tags, as unit tests rely on.
/// </remarks>
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
StopTimers();
if (HasConfigBody(driverConfigJson)) AdoptConfig(driverConfigJson);
_running = true;
StartTimers();
_logger.LogInformation(
@@ -279,18 +317,8 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba
// 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)
{
// Queued evaluations were prepared against the OUTGOING tag table (old script source, old
// dependency set), so they are stale the moment the rebuild lands — drop them rather than
// publish one last pre-edit value after the delta was applied.
_pending.Clear();
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();
AdoptConfig(driverConfigJson);
// Re-arm timers off the rebuilt tag set.
_running = true;
StartTimers();
return Task.CompletedTask;
@@ -348,9 +348,18 @@ public sealed class GalaxyDriver
ReplayAsync(cancellationToken);
/// <inheritdoc />
/// <remarks>
/// #664: <c>InitializeAsync</c> is reachable on a live driver with a NEW config —
/// <c>DriverInstanceActor.AdoptConfigDuringInit</c> re-runs it when an ApplyDelta lands mid-connect. The
/// driver cannot hot-swap its options (they shape the gw session; see <see cref="ReinitializeAsync"/>), so a
/// non-equivalent config with a real body is REFUSED here too rather than silently ignored — the driver
/// would otherwise connect on its old endpoint while the deploy reported the change applied. A
/// <c>"{}"</c> / empty document is the unit-test lifecycle shape and keeps the constructor options.
/// </remarks>
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
ObjectDisposedException.ThrowIf(_disposed, this);
if (HasConfigBody(driverConfigJson)) RefuseNonEquivalentConfig(driverConfigJson);
// Tests inject seams via the internal ctor; production InitializeAsync builds
// the gateway client + session + per-capability runtime components from
@@ -650,7 +659,27 @@ public sealed class GalaxyDriver
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
ObjectDisposedException.ThrowIf(_disposed, this);
if (!string.IsNullOrWhiteSpace(driverConfigJson))
if (!string.IsNullOrWhiteSpace(driverConfigJson)) RefuseNonEquivalentConfig(driverConfigJson);
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
return Task.CompletedTask;
}
/// <summary>True when the supplied DriverConfig JSON carries a real body (not empty / <c>"{}"</c> / <c>"[]"</c>).</summary>
private static bool HasConfigBody(string? driverConfigJson)
{
if (string.IsNullOrWhiteSpace(driverConfigJson)) return false;
var trimmed = driverConfigJson.Trim();
return trimmed is not "{}" and not "[]";
}
/// <summary>
/// Throw <see cref="NotSupportedException"/> unless <paramref name="driverConfigJson"/> describes options
/// runtime-equivalent to the live ones (see <see cref="OptionsAreEquivalent"/>). Shared by
/// <see cref="ReinitializeAsync"/> and <see cref="InitializeAsync"/> (#664) so neither path can silently
/// ignore a config change it cannot apply.
/// </summary>
private void RefuseNonEquivalentConfig(string driverConfigJson)
{
// In-place config reapply. The driver does not currently support hot-swapping
// GalaxyDriverOptions at runtime — changing the gateway endpoint, MxAccess
@@ -676,21 +705,18 @@ public sealed class GalaxyDriver
catch (Exception ex) when (ex is not NotSupportedException and not ObjectDisposedException)
{
throw new NotSupportedException(
$"GalaxyDriver.ReinitializeAsync could not parse the incoming DriverConfig JSON for '{_driverInstanceId}': {ex.Message}",
$"GalaxyDriver could not parse the incoming DriverConfig JSON for '{_driverInstanceId}': {ex.Message}",
ex);
}
if (!OptionsAreEquivalent(_options, incoming))
{
throw new NotSupportedException(
"GalaxyDriver.ReinitializeAsync does not support hot-swapping driver options at runtime " +
"GalaxyDriver does not support hot-swapping driver options at runtime " +
"(gateway endpoint, MxAccess client name, reconnect policy, etc.). Restart the DriverInstance " +
"through the host stack to apply a config change.");
}
}
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
return Task.CompletedTask;
}
/// <summary>
/// Compare two <see cref="GalaxyDriverOptions"/> for runtime equivalence — every
@@ -221,6 +221,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
private readonly ILoggingAdapter _log = Context.GetLogger();
private string? _currentConfigJson;
/// <summary>#664: set when <see cref="AdoptConfigDuringInit"/> adopts a new config mid-connect; the next
/// (generation-matched) <see cref="InitializeSucceeded"/> sends <see cref="DeltaApplied"/> to the parent and
/// clears it, so a dependency consumer's mux interest is re-registered against the ADOPTED config — the
/// same post-reinit notification the Connected <see cref="HandleApplyDeltaAsync"/> path sends.</summary>
private bool _adoptedConfigPendingNotify;
/// <summary>Monotonic token tagging each <see cref="InitializeAsync"/> attempt. An init result is
/// honoured only when its generation matches the latest; an older result is from a superseded attempt
/// (e.g. an <see cref="ApplyDelta"/> adopted a new config mid-(re)connect) and is dropped. Touched only
@@ -468,6 +474,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
if (msg.Generation != _initGeneration) return;
_log.Info("DriverInstance {Id}: connected", _driverInstanceId);
Become(Connected);
NotifyAdoptedConfig();
PublishHealthSnapshot();
ResubscribeDesired();
AttachAlarmSource();
@@ -771,6 +778,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Timers.Cancel("retry-connect");
_log.Info("DriverInstance {Id}: reconnected", _driverInstanceId);
Become(Connected);
NotifyAdoptedConfig();
PublishHealthSnapshot();
ResubscribeDesired();
AttachAlarmSource();
@@ -878,9 +886,19 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
_log.Info("DriverInstance {Id}: ApplyDelta during (re)connect — adopting new config, re-initialising now",
_driverInstanceId);
InitializeAsync(msg.DriverConfigJson);
_adoptedConfigPendingNotify = true;
Sender.Tell(new ApplyResult(true, "config adopted; reinitializing", msg.Correlation));
}
/// <summary>Send the deferred <see cref="DeltaApplied"/> for a mid-connect adoption (#664) once the driver
/// has actually initialised on the adopted config. A no-op otherwise.</summary>
private void NotifyAdoptedConfig()
{
if (!_adoptedConfigPendingNotify) return;
_adoptedConfigPendingNotify = false;
Context.Parent.Tell(new DeltaApplied(_driverInstanceId));
}
/// <summary>Adopt a new config while Connected, in place, via <see cref="IDriver.ReinitializeAsync"/>.
/// <para>#663: every driver's reinit releases its device/session state (#630), so the driver-side data and
/// native-alarm subscriptions do not survive it — but this handler used to keep both handles, and the
@@ -413,6 +413,34 @@ public sealed class CalculationDriverTests
report.SkippedCount.ShouldBe(1, "the reinit payload's tag has no scriptId, so it is no longer a calc tag");
}
/// <summary>
/// Gitea #664 — <c>InitializeAsync</c> re-runs on a LIVE driver with a new config when an ApplyDelta lands
/// mid-connect (<c>DriverInstanceActor.AdoptConfigDuringInit</c>). It used to ignore the JSON, so the
/// driver kept its constructor tag set while the deploy reported the change applied. A config with a body
/// now rebuilds the tag table (and the dependency-ref set the host re-registers from).
/// </summary>
[Fact]
public async Task InitializeAsync_on_a_live_driver_adopts_a_config_with_a_body()
{
var (driver, published, _) = Build(CalcTag("calc/e/a", "return (double)ctx.GetTag(\"src/a\").Value * 2.0;"));
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
driver.DependencyRefs.ShouldBe(new[] { "src/a" });
var adopted = System.Text.Json.JsonSerializer.Serialize(new
{
rawTags = new[] { CalcTag("calc/e/b", "return (double)ctx.GetTag(\"src/b\").Value + 1.0;") },
});
await driver.InitializeAsync(adopted, TestContext.Current.CancellationToken);
driver.DependencyRefs.ShouldBe(new[] { "src/b" });
driver.OnDependencyValue("src/b", 4.0, StatusGood, DateTime.UtcNow);
await Settle(driver);
published.ShouldContain(e => e.FullReference == "calc/e/b" && Equals(e.Snapshot.Value, 5.0));
published.ShouldNotContain(e => e.FullReference == "calc/e/a");
await driver.ShutdownAsync(TestContext.Current.CancellationToken);
}
/// <summary>Minimal capturing <see cref="ILogger{T}"/> so a test can assert on what the driver's own
/// logger received (mirrors the equivalent helper in CalculationEvaluatorTests).</summary>
private sealed class CapturingLogger : ILogger<CalculationDriver>
@@ -168,7 +168,9 @@ public sealed class GalaxyDriverFactoryTests
dataWriter: null, subscriber: new NoopSubscriber());
driver.GetHealth().State.ShouldBe(DriverState.Unknown);
await driver.InitializeAsync(MinimalConfig, CancellationToken.None);
// #664: InitializeAsync now refuses a config that differs from the live options, so this uses the
// "{}" lifecycle shape (MinimalConfig's ApiKeySecretRef differs from BuildOptions()).
await driver.InitializeAsync("{}", CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
@@ -206,6 +208,38 @@ public sealed class GalaxyDriverFactoryTests
driver.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThan(firstStamp);
}
/// <summary>
/// Gitea #664 — <c>InitializeAsync</c> re-runs on a live driver with a NEW config when an ApplyDelta lands
/// mid-connect. Galaxy cannot hot-swap its options, and it used to ignore the JSON — connecting on the OLD
/// endpoint while the deploy reported the change applied. A non-equivalent config is now refused, exactly
/// as <c>ReinitializeAsync</c> refuses it; an equivalent one still initialises.
/// </summary>
[Fact]
public async Task InitializeAsync_refuses_a_non_equivalent_config_and_accepts_an_equivalent_one()
{
const string moved = """
{
"Gateway": { "Endpoint": "https://elsewhere.test:5001", "ApiKeySecretRef": "key" },
"MxAccess": { "ClientName": "OtOpcUa-A" }
}
""";
const string same = """
{
"Gateway": { "Endpoint": "https://mxgw.test:5001", "ApiKeySecretRef": "key" },
"MxAccess": { "ClientName": "OtOpcUa-A" }
}
""";
using var driver = new GalaxyDriver(
"galaxy-x", BuildOptions(), hierarchySource: null, dataReader: null,
dataWriter: null, subscriber: new NoopSubscriber());
await Should.ThrowAsync<NotSupportedException>(() => driver.InitializeAsync(moved, CancellationToken.None));
driver.GetHealth().State.ShouldBe(DriverState.Unknown, "a refused init must not report Healthy");
await driver.InitializeAsync(same, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
}
private static GalaxyDriverOptions BuildOptions() => new(
new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
new GalaxyMxAccessOptions("OtOpcUa-A"),
@@ -283,7 +317,8 @@ public sealed class GalaxyDriverFactoryTests
using var driver = new GalaxyDriver(
"g", BuildOptions(), hierarchySource: null, dataReader: null,
dataWriter: null, subscriber: new NoopSubscriber());
await driver.InitializeAsync(MinimalConfig, CancellationToken.None);
// #664: "{}" — MinimalConfig is not equivalent to BuildOptions(), and InitializeAsync now refuses that.
await driver.InitializeAsync("{}", CancellationToken.None);
driver.GetHostStatuses().ShouldBeEmpty();
}
@@ -338,6 +338,40 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
driver.InitConfigs.ShouldContain(good);
}
/// <summary>
/// Gitea #664 — a mid-connect adoption sends <see cref="DriverInstanceActor.DeltaApplied"/> once the
/// driver has initialised on the ADOPTED config, so the host re-registers a dependency consumer's mux
/// interest (the Calculation driver's refs change with its tag set). It used to send nothing, leaving the
/// old interest registered. The notification is one-shot: a later reconnect on the same config sends none.
/// </summary>
[Fact]
public void A_mid_connect_adoption_notifies_DeltaApplied_once_after_the_adopted_init_succeeds()
{
const string bad = "{\"v\":\"bad\"}";
const string good = "{\"v\":\"good\"}";
var driver = new SubscribableStubDriver
{
InitBehavior = cfg => cfg == good ? Task.CompletedTask : throw new InvalidOperationException("bad-cfg"),
};
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is not DriverInstanceActor.DeltaApplied);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, reconnectInterval: TimeSpan.FromMilliseconds(50)));
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)));
actor.Tell(new DriverInstanceActor.InitializeRequested(bad));
AwaitCondition(() => driver.InitializeCount >= 2, PresenceBudget);
actor.Tell(new DriverInstanceActor.ApplyDelta(good, CorrelationId.NewId()));
parent.ExpectMsg<DriverInstanceActor.DeltaApplied>(PresenceBudget);
driver.InitConfigs.ShouldContain(good, "the notification must follow an init on the adopted config");
// A reconnect on the same config re-runs InitializeSucceeded; the positive control is its re-subscribe.
var subscribes = driver.SubscribeCount;
actor.Tell(new DriverInstanceActor.ForceReconnect());
AwaitCondition(() => driver.SubscribeCount > subscribes, PresenceBudget);
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
/// <summary>A stale InitializeSucceeded from an old (superseded) config cannot hijack the state:
/// while a gated old-config init is pending in Connecting, an ApplyDelta adopts a new config; the
/// old init completing afterwards is ignored, and only the new config drives Connected.</summary>