From 97fcf624aeab35910d39778de909b2a26dd1681b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 11:05:58 -0400 Subject: [PATCH] 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 --- CLAUDE.md | 3 + docs/DriverLifecycle.md | 10 ++ .../CalculationDriver.cs | 52 ++++++--- .../GalaxyDriver.cs | 102 +++++++++++------- .../Drivers/DriverInstanceActor.cs | 18 ++++ .../CalculationDriverTests.cs | 28 +++++ .../GalaxyDriverFactoryTests.cs | 39 ++++++- .../Drivers/DriverInstanceActorTests.cs | 34 ++++++ 8 files changed, 234 insertions(+), 52 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 48c557c9..9a9e0135 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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. diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index dda21ba4..ca39046f 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -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 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 99ee773b..af48df42 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs @@ -239,6 +239,35 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba return new CalculationDriverOptions { RawTags = _options.RawTags, RunTimeout = _options.RunTimeout }; } + /// Re-bind the tag table from — shared by + /// and (for a config with a body) so the two + /// adoption paths cannot drift. Timers must already be stopped. + 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(); + } + + /// True when the supplied DriverConfig JSON carries a real body. The bootstrapper always passes a + /// populated document; unit tests pass "{}" to exercise lifecycle shape — those keep the + /// constructor-supplied tags (an empty document would otherwise bind to an EMPTY tag table). + private static bool HasConfigBody(string? driverConfigJson) + { + if (string.IsNullOrWhiteSpace(driverConfigJson)) return false; + var trimmed = driverConfigJson.Trim(); + return trimmed is not "{}" and not "[]"; + } + /// 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. @@ -257,8 +286,17 @@ public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscriba public string DriverType => DriverTypeNames.Calculation; /// + /// + /// #664: InitializeAsync is reachable on a LIVE driver — DriverInstanceActor.AdoptConfigDuringInit + /// 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 "{}" / empty + /// document keeps the constructor-supplied tags, as unit tests rely on. + /// 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; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index f464a816..693619e0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -348,9 +348,18 @@ public sealed class GalaxyDriver ReplayAsync(cancellationToken); /// + /// + /// #664: InitializeAsync is reachable on a live driver with a NEW config — + /// DriverInstanceActor.AdoptConfigDuringInit re-runs it when an ApplyDelta lands mid-connect. The + /// driver cannot hot-swap its options (they shape the gw session; see ), 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 + /// "{}" / empty document is the unit-test lifecycle shape and keeps the constructor options. + /// 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,48 +659,65 @@ public sealed class GalaxyDriver public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { ObjectDisposedException.ThrowIf(_disposed, this); - if (!string.IsNullOrWhiteSpace(driverConfigJson)) - { - // In-place config reapply. The driver does not currently support hot-swapping - // GalaxyDriverOptions at runtime — changing the gateway endpoint, MxAccess - // client name, or reconnect policy requires tearing down the gw session, - // supervisor, event pump, and address space. The host stack handles that via - // DriverInstance restart, so this method only accepts an equivalent config (no - // meaningful change) and refreshes health; a non-equivalent reapply throws - // NotSupportedException so the caller knows the change wasn't applied. - // - // Materialise the incoming config and compare against the live options. We - // refuse any change that would require a session teardown rather than - // pretending to apply it. - GalaxyDriverOptions incoming; - try - { - // Reuse the factory's parse pipeline so any missing-required-field - // error surfaces with the same diagnostic text as InitializeAsync. - var transient = GalaxyDriverFactoryExtensions.CreateInstance(_driverInstanceId, driverConfigJson); - incoming = transient.Options; - // The transient instance never started a runtime — disposing is cheap. - transient.Dispose(); - } - 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}", - ex); - } - - if (!OptionsAreEquivalent(_options, incoming)) - { - throw new NotSupportedException( - "GalaxyDriver.ReinitializeAsync 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."); - } - } + if (!string.IsNullOrWhiteSpace(driverConfigJson)) RefuseNonEquivalentConfig(driverConfigJson); _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); return Task.CompletedTask; } + + /// True when the supplied DriverConfig JSON carries a real body (not empty / "{}" / "[]"). + private static bool HasConfigBody(string? driverConfigJson) + { + if (string.IsNullOrWhiteSpace(driverConfigJson)) return false; + var trimmed = driverConfigJson.Trim(); + return trimmed is not "{}" and not "[]"; + } + + /// + /// Throw unless describes options + /// runtime-equivalent to the live ones (see ). Shared by + /// and (#664) so neither path can silently + /// ignore a config change it cannot apply. + /// + 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 + // client name, or reconnect policy requires tearing down the gw session, + // supervisor, event pump, and address space. The host stack handles that via + // DriverInstance restart, so this method only accepts an equivalent config (no + // meaningful change) and refreshes health; a non-equivalent reapply throws + // NotSupportedException so the caller knows the change wasn't applied. + // + // Materialise the incoming config and compare against the live options. We + // refuse any change that would require a session teardown rather than + // pretending to apply it. + GalaxyDriverOptions incoming; + try + { + // Reuse the factory's parse pipeline so any missing-required-field + // error surfaces with the same diagnostic text as InitializeAsync. + var transient = GalaxyDriverFactoryExtensions.CreateInstance(_driverInstanceId, driverConfigJson); + incoming = transient.Options; + // The transient instance never started a runtime — disposing is cheap. + transient.Dispose(); + } + catch (Exception ex) when (ex is not NotSupportedException and not ObjectDisposedException) + { + throw new NotSupportedException( + $"GalaxyDriver could not parse the incoming DriverConfig JSON for '{_driverInstanceId}': {ex.Message}", + ex); + } + + if (!OptionsAreEquivalent(_options, incoming)) + { + throw new NotSupportedException( + "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."); + } + } + /// /// Compare two for runtime equivalence — every /// field that drives gw session shape or reconnect behaviour must match. Compared 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 dd669688..ad3de17c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -221,6 +221,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private readonly ILoggingAdapter _log = Context.GetLogger(); private string? _currentConfigJson; + /// #664: set when adopts a new config mid-connect; the next + /// (generation-matched) sends 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 path sends. + private bool _adoptedConfigPendingNotify; + /// Monotonic token tagging each attempt. An init result is /// honoured only when its generation matches the latest; an older result is from a superseded attempt /// (e.g. an 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)); } + /// Send the deferred for a mid-connect adoption (#664) once the driver + /// has actually initialised on the adopted config. A no-op otherwise. + private void NotifyAdoptedConfig() + { + if (!_adoptedConfigPendingNotify) return; + _adoptedConfigPendingNotify = false; + Context.Parent.Tell(new DeltaApplied(_driverInstanceId)); + } + /// Adopt a new config while Connected, in place, via . /// #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 diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs index 42e11132..d9460d99 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation.Tests/CalculationDriverTests.cs @@ -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"); } + /// + /// Gitea #664 — InitializeAsync re-runs on a LIVE driver with a new config when an ApplyDelta lands + /// mid-connect (DriverInstanceActor.AdoptConfigDuringInit). 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). + /// + [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); + } + /// Minimal capturing so a test can assert on what the driver's own /// logger received (mirrors the equivalent helper in CalculationEvaluatorTests). private sealed class CapturingLogger : ILogger diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverFactoryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverFactoryTests.cs index 875dde8d..91954ab5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverFactoryTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverFactoryTests.cs @@ -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); } + /// + /// Gitea #664 — InitializeAsync 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 ReinitializeAsync refuses it; an equivalent one still initialises. + /// + [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(() => 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(); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs index 9dd9eb0c..9bb72614 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs @@ -338,6 +338,40 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase driver.InitConfigs.ShouldContain(good); } + /// + /// Gitea #664 — a mid-connect adoption sends 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. + /// + [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(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)); + } + /// 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.