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

refactor(runtime): delete the dead in-place ApplyDelta seam; config change = respawn only (#665)

Browse Source 
Since #516 a DriverConfig change is stop + respawn, so DriverSpawnPlan.ToApplyDelta
was always empty and the in-place path had no producer. #663 and #664 were both
latent bugs in it. This deletes it:
- host: ApplyChildDelta, _pendingDelta, HandleApplyResult and HandleDeltaApplied
- plan: DriverSpawnPlan.ToApplyDelta
- actor: the ApplyDelta, ApplyResult and DeltaApplied messages, the Connected
  reinit handler and the mid-connect AdoptConfigDuringInit
- tests: the actor tests that drove the seam

IDriver.ReinitializeAsync and each driver's implementation and unit tests stay.
Comments and docs that cited AdoptConfigDuringInit as the reason a driver must
release prior state at init now cite the IDriver contract and overlapping
retries, which still apply.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 11:23:16 -04:00
parent f622d0be2d
commit 1ea1da8236
33 changed files with 112 additions and 559 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs
+5 -4
View File
@@ -31,7 +31,7 @@ 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
/// <summary>Build the merged DriverConfig JSON an in-place re-bind reads (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 });
@@ -130,7 +130,7 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase
published.ShouldContain(e => e.FullReference == "calc/remove" && Equals(e.Snapshot.Value, 7.0));
}, PresenceBudget, TimeSpan.FromMilliseconds(100));
// ---- REDEPLOY (in-place ApplyDelta) ----
// ---- IN-PLACE RE-BIND (driver ReinitializeAsync) ----
// 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(
@@ -141,8 +141,9 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase
// 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).
// The adapter re-reads the driver's refs only when told to; re-register AFTER the reinit so it adopts the
// NEW ref set (incl. the brand-new src/b). (The runtime respawns on a config change, #665, so a fresh
// adapter registers the new set in PreStart; this exercises the driver + adapter contract directly.)
adapter.Tell(new DependencyConsumerMuxAdapter.ReRegister());
// Leg 2 — after redeploy: the NEW dep src/b flows + the NEW tag computes; the EDITED script serves the
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs
+8 -104
View File
@@ -369,56 +369,6 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
.Connected.ShouldBeTrue();
}
/// <summary>
/// Gitea #663 — an in-place config change on a Connected driver (<see cref="DriverInstanceActor.ApplyDelta"/>
/// → <see cref="IDriver.ReinitializeAsync"/>) must end with exactly one live data subscription and one live
/// native-alarm subscription at the driver, and the alarm feed still forwarding. The handler used to keep
/// the actor's handles across the reinit while the driver had dropped the subscriptions behind them, so
/// the reconcile (which only re-subscribes a MISSING handle) never restored them and the driver went silent.
/// </summary>
/// <param name="reinitKeepsSubscriptions">false = the production shape (reinit drops everything, #630) —
/// the regression; true = a driver that keeps them, which pins that the fix releases before re-subscribing
/// instead of piling a second subscription on top.</param>
[Theory]
[InlineData(false)]
[InlineData(true)]
public void An_in_place_ApplyDelta_leaves_one_live_data_and_one_live_alarm_subscription(bool reinitKeepsSubscriptions)
{
var driver = new AlarmSourceStubDriver { ReinitKeepsSubscriptions = reinitKeepsSubscriptions };
var publisher = new CountingHealthPublisher();
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
driver, healthPublisher: publisher, healthPollInterval: TimeSpan.FromMilliseconds(100)));
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
new[] { "tag-a" }, TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" }));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.LiveDataSubscriptions == 1 && driver.LiveAlarmSubscriptions == 1, PresenceBudget);
var correlation = CorrelationId.NewId();
actor.Tell(new DriverInstanceActor.ApplyDelta("{\"changed\":true}", correlation), parent.Ref);
parent.FishForMessage<DriverInstanceActor.ApplyResult>(r => r.Correlation == correlation, PresenceBudget)
.Success.ShouldBeTrue();
driver.ReinitializeCount.ShouldBe(1);
driver.InitializeCount.ShouldBe(1, "an in-place delta must not go through the reconnect path");
AwaitCondition(() => driver.LiveDataSubscriptions == 1 && driver.LiveAlarmSubscriptions == 1, PresenceBudget);
// The "not two" half is an absence, so wait for three more health-poll ticks — each runs the reconcile
// that would issue a duplicate — before re-checking.
var ticks = publisher.Count;
AwaitCondition(() => publisher.Count >= ticks + 3, PresenceBudget);
driver.LiveDataSubscriptions.ShouldBe(1);
driver.LiveAlarmSubscriptions.ShouldBe(1);
driver.AlarmSubscriberCount.ShouldBe(1);
driver.RaiseAlarm(new AlarmEventArgs(
new StubAlarmHandle(), "Temp.HiHi", "Temp.HiHi", "T", "after reinit", AlarmSeverity.High,
DateTime.UtcNow, Kind: AlarmTransitionKind.Raise));
parent.FishForMessage<DriverInstanceActor.AttributeAlarmPublished>(_ => true, PresenceBudget)
.Args.ConditionId.ShouldBe("Temp.HiHi");
}
// --- stub drivers ----------------------------------------------------------------------------
private class StubDriver : IDriver
@@ -439,7 +389,7 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
}
/// <summary>Reinitializes the driver.</summary>
public virtual Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) =>
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) =>
Task.CompletedTask;
/// <summary>Shuts down the driver.</summary>
@@ -463,55 +413,14 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
/// <summary>Number of live subscribers on <see cref="OnAlarmEvent"/> — proves attach/detach.</summary>
public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0;
private readonly object _liveLock = new();
private readonly HashSet<object> _liveData = new(ReferenceEqualityComparer.Instance);
private readonly HashSet<object> _liveAlarms = new(ReferenceEqualityComparer.Instance);
private int _reinitializeCount;
/// <summary>Driver-side data subscriptions currently open (subscribed and not yet released).</summary>
public int LiveDataSubscriptions { get { lock (_liveLock) return _liveData.Count; } }
/// <summary>Driver-side native-alarm subscriptions currently open.</summary>
public int LiveAlarmSubscriptions { get { lock (_liveLock) return _liveAlarms.Count; } }
/// <summary>Number of <see cref="ReinitializeAsync"/> calls.</summary>
public int ReinitializeCount => Volatile.Read(ref _reinitializeCount);
/// <summary>When false (the default, and every production driver since #630), a reinit releases the
/// device/session state and with it every driver-side subscription; when true it keeps them — the
/// shape that would double-subscribe if the actor re-subscribed without releasing first.</summary>
public bool ReinitKeepsSubscriptions { get; init; }
/// <inheritdoc />
public override Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
Interlocked.Increment(ref _reinitializeCount);
if (!ReinitKeepsSubscriptions)
{
lock (_liveLock)
{
_liveData.Clear();
_liveAlarms.Clear();
}
}
return Task.CompletedTask;
}
/// <summary>Subscribes to the specified full references.</summary>
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
var handle = new StubHandle();
lock (_liveLock) _liveData.Add(handle);
return Task.FromResult<ISubscriptionHandle>(handle);
}
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
Task.FromResult<ISubscriptionHandle>(_subHandle);
/// <summary>Unsubscribes from the specified subscription handle.</summary>
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
lock (_liveLock) _liveData.Remove(handle);
return Task.CompletedTask;
}
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) =>
Task.CompletedTask;
private int _subscribeAlarmsCallCount;
@@ -530,9 +439,7 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
var call = Interlocked.Increment(ref _subscribeAlarmsCallCount);
if (call <= AlarmSubscribeFailuresBeforeSuccess)
throw new InvalidOperationException($"alarm subscribe failure {call} (test-injected)");
var handle = new StubAlarmHandle();
lock (_liveLock) _liveAlarms.Add(handle);
return Task.FromResult<IAlarmSubscriptionHandle>(handle);
return Task.FromResult<IAlarmSubscriptionHandle>(new StubAlarmHandle());
}
/// <summary>How many leading <see cref="SubscribeAlarmsAsync"/> calls throw — the "session not ready
@@ -540,11 +447,8 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
public int AlarmSubscribeFailuresBeforeSuccess { get; init; }
/// <summary>Cancels an alarm subscription.</summary>
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
{
lock (_liveLock) _liveAlarms.Remove(handle);
return Task.CompletedTask;
}
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) =>
Task.CompletedTask;
/// <summary>Acknowledges a batch of alarms.</summary>
public Task AcknowledgeAsync(
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorTests.cs
-125
View File
@@ -11,28 +11,6 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
public sealed class DriverInstanceActorTests : RuntimeActorTestBase
{
/// <summary>Verifies that ApplyDelta calls ReinitializeAsync when connected and replies success.</summary>
[Fact]
public async Task ApplyDelta_when_Connected_calls_ReinitializeAsync_and_replies_success()
{
var driver = new StubDriver();
var actor = Sys.ActorOf(DriverInstanceActor.Props(driver));
// Drive: Initialize → Connected.
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.InitializeCount > 0, TimeSpan.FromSeconds(2));
// Issue ApplyDelta and capture the reply via Ask.
var correlation = CorrelationId.NewId();
var reply = await actor.Ask<DriverInstanceActor.ApplyResult>(
new DriverInstanceActor.ApplyDelta("{\"changed\":true}", correlation),
TimeSpan.FromSeconds(3));
reply.Success.ShouldBeTrue();
reply.Correlation.ShouldBe(correlation);
driver.ReinitializeCount.ShouldBe(1);
}
/// <summary>Verifies that initialize failure keeps the actor in Reconnecting state.</summary>
[Fact]
public void Initialize_failure_keeps_actor_in_Reconnecting_state()
@@ -305,109 +283,6 @@ public sealed class DriverInstanceActorTests : RuntimeActorTestBase
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
/// <summary>A driver stuck Reconnecting (init failing on a bad config) adopts a corrected config
/// delivered via ApplyDelta and connects on it — no node restart. Closes pending.md #7.</summary>
[Fact]
public async Task ApplyDelta_while_Reconnecting_adopts_new_config_and_connects()
{
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();
// #477: DriverInstanceActor now Tells the parent ConnectivityChanged on each connect/
// disconnect; these tests assert on the data/alarm/discovery forwards, so ignore it.
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
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, TimeSpan.FromSeconds(2));
driver.SubscribeCount.ShouldBe(0);
var correlation = CorrelationId.NewId();
var reply = await actor.Ask<DriverInstanceActor.ApplyResult>(
new DriverInstanceActor.ApplyDelta(good, correlation), TimeSpan.FromSeconds(3));
reply.Success.ShouldBeTrue();
reply.Correlation.ShouldBe(correlation);
AwaitCondition(() => driver.SubscribeCount >= 1, TimeSpan.FromSeconds(3));
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>
[Fact]
public async Task ApplyDelta_supersedes_in_flight_init_so_stale_result_is_ignored()
{
const string v1 = "{\"v\":1}";
const string v2 = "{\"v\":2}";
var gate1 = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var gate2 = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var driver = new SubscribableStubDriver
{
InitBehavior = cfg => cfg == v1 ? gate1.Task : gate2.Task,
};
var parent = CreateTestProbe();
// #477: DriverInstanceActor now Tells the parent ConnectivityChanged on each connect/
// disconnect; these tests assert on the data/alarm/discovery forwards, so ignore it.
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, reconnectInterval: TimeSpan.FromSeconds(30)));
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)));
actor.Tell(new DriverInstanceActor.InitializeRequested(v1));
AwaitCondition(() => driver.InitConfigs.Contains(v1), TimeSpan.FromSeconds(2));
await actor.Ask<DriverInstanceActor.ApplyResult>(
new DriverInstanceActor.ApplyDelta(v2, CorrelationId.NewId()), TimeSpan.FromSeconds(3));
AwaitCondition(() => driver.InitConfigs.Contains(v2), TimeSpan.FromSeconds(2));
gate1.SetResult();
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
driver.SubscribeCount.ShouldBe(0);
gate2.SetResult();
AwaitCondition(() => driver.SubscribeCount >= 1, TimeSpan.FromSeconds(3));
}
/// <summary>
/// Archreview <b>03/S15 (T11)</b> — a child that is <b>stopped</b> announces comms loss on its way
/// out, so the host latches its value nodes <c>BadNoCommunication</c>.
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverSpawnPlannerTests.cs
-10
View File
@@ -19,7 +19,6 @@ public sealed class DriverSpawnPlannerTests
var plan = DriverSpawnPlanner.Compute(current, target);
plan.ToSpawn.Count.ShouldBe(2);
plan.ToApplyDelta.ShouldBeEmpty();
plan.ToStop.ShouldBeEmpty();
}
@@ -36,7 +35,6 @@ public sealed class DriverSpawnPlannerTests
var plan = DriverSpawnPlanner.Compute(current, target);
plan.ToSpawn.ShouldBeEmpty();
plan.ToApplyDelta.ShouldBeEmpty();
plan.ToStop.ShouldBeEmpty();
}
@@ -65,7 +63,6 @@ public sealed class DriverSpawnPlannerTests
plan.ToSpawn.Single().DriverInstanceId.ShouldBe("a");
plan.ToSpawn.Single().DriverConfig.ShouldBe("{\"host\":\"new\"}");
// The whole point: nothing is left on the in-place path that could discard the edit.
plan.ToApplyDelta.ShouldBeEmpty();
}
/// <summary>Verifies that removed drivers are routed to ToStop.</summary>
@@ -83,7 +80,6 @@ public sealed class DriverSpawnPlannerTests
plan.ToStop.ShouldBe(new[] { "b" });
plan.ToSpawn.ShouldBeEmpty();
plan.ToApplyDelta.ShouldBeEmpty();
}
/// <summary>Verifies that disabled drivers with running children are routed to ToStop.</summary>
@@ -100,7 +96,6 @@ public sealed class DriverSpawnPlannerTests
plan.ToStop.Single().ShouldBe("a");
plan.ToSpawn.ShouldBeEmpty();
plan.ToApplyDelta.ShouldBeEmpty();
}
/// <summary>Verifies that disabled new drivers are not spawned.</summary>
@@ -113,7 +108,6 @@ public sealed class DriverSpawnPlannerTests
var plan = DriverSpawnPlanner.Compute(current, target);
plan.ToSpawn.ShouldBeEmpty();
plan.ToApplyDelta.ShouldBeEmpty();
plan.ToStop.ShouldBeEmpty();
}
@@ -131,7 +125,6 @@ public sealed class DriverSpawnPlannerTests
plan.ToStop.Single().ShouldBe("a");
plan.ToSpawn.Single().DriverType.ShouldBe("AbCip");
plan.ToApplyDelta.ShouldBeEmpty();
}
/// <summary>
@@ -153,7 +146,6 @@ public sealed class DriverSpawnPlannerTests
plan.ToStop.Single().ShouldBe("a");
plan.ToSpawn.Single().DriverInstanceId.ShouldBe("a");
plan.ToApplyDelta.ShouldBeEmpty();
}
/// <summary>
@@ -173,7 +165,6 @@ public sealed class DriverSpawnPlannerTests
plan.ToStop.Single().ShouldBe("a");
plan.ToSpawn.Single().DriverInstanceId.ShouldBe("a");
plan.ToApplyDelta.ShouldBeEmpty();
}
/// <summary>
@@ -196,6 +187,5 @@ public sealed class DriverSpawnPlannerTests
plan.ToStop.Single().ShouldBe("a");
plan.ToSpawn.Single().DriverInstanceId.ShouldBe("a");
plan.ToApplyDelta.ShouldBeEmpty();
}
}