fix(runtime): rate-limit the supervision reprovision — deferred, not dropped (review F3)

The default decider restarts a faulting child immediately and without limit, and
the hook makes every restart do ShutdownAsync+InitializeAsync+SubscribeAsync, so a
driver faulting on every value became a reconnect storm. Per-driver last-reprovision
stamp + DefaultReprovisionBackoff (10s, matched to DefaultReconnectInterval);
a held request is re-delivered by a single per-driver timer carrying the original
incarnation ref, so the F2 identity check still applies.

Seen RED with the backoff disabled: InitializeCount == 3 (second applied
immediately) where the test requires 2-then-3.

Also: the _children.Keys union-arm test (seeder gained taglessDrivers) and the
_desiredByDriver xmldoc note that it lingers on last-known-good after a failed apply.
This commit is contained in:
Joseph Doherty
2026-08-20 10:39:13 -04:00
parent 2b7100cf19
commit 0793b0cba8
4 changed files with 212 additions and 13 deletions
+18
View File
@@ -201,6 +201,24 @@ Things to know before touching any of this:
remove a driver; a bounded retry count would silently stop one after N crashes. Both change failure
semantics for every driver in the fleet and deserve their own plan and live gate. The reprovision hook
makes the *current* infinite-restart behaviour correct instead of dead.
- ⚠️ **…which is exactly why the reprovision is RATE-LIMITED.** The default decider restarts a faulting
child **immediately and without limit**, and the hook now makes every one of those restarts do
`ShutdownAsync` + `InitializeAsync` + `SubscribeAsync` — so a driver that faults on every value would
hammer its device as fast as it can crash. `DriverHostActor` keeps a per-driver last-reprovision stamp
and holds anything inside `DefaultReprovisionBackoff` (10 s, matched to
`DriverInstanceActor.DefaultReconnectInterval` because it drives the same connect work). The held
request is **deferred, not dropped** — one `StartSingleTimer` per driver, so a burst collapses to a
single pending reprovision — because dropping it would leave a child whose *final* crash fell inside
the window un-provisioned forever, which is the 03/S16 failure this hook exists to close. Each
deferral logs a Warning naming the crash type, so a crash loop is visible rather than merely quiet.
- **A reprovision is routed by INCARNATION, not by driver id.** `ReprovisionRequested` is delivered
asynchronously, so one can still be in flight from an old incarnation after `HandleRestartDriver`
stopped that child and spawned a replacement under the same id. Delivered to the new child — which is
in `Connecting`, where `InitializeRequested` *is* handled — it would start a second, concurrent
`InitializeAsync`, racing the dispose-before-overwrite that #527's teardown exists to serialise
(`ModbusDriver._transport` / `MqttDriver._connection` each leak a transport on that race). The handler
therefore checks `Sender` against the child it currently holds and drops anything else; the deferred
path carries the original ref so the same check still applies when the timer re-delivers it.
- **Ordering on a respawning deploy, and its one accepted cost.** `ApplyDriverSpecs` (which spawns) runs
before the pass that rebuilds `_desiredByDriver`, so a child respawned by a deploy — raw tags are folded
into the merged `DriverConfig`, so a tag-set edit changes the spec and trips the #516 config respawn —
@@ -82,6 +82,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// </remarks>
public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5);
/// <summary>Minimum interval between two supervision reprovisions of the SAME driver child (archreview
/// review F3). Akka's default decider restarts a faulting child immediately and without limit, and the
/// reprovision hook makes every restart do ShutdownAsync + InitializeAsync + SubscribeAsync — so an
/// un-limited hook turns a driver that faults on every value into a reconnect storm against its device.
/// Matched to <see cref="DriverInstanceActor.DefaultReconnectInterval"/>: the reprovision drives the same
/// connect work, so it should not run more often than an ordinary reconnect attempt.</summary>
public static readonly TimeSpan DefaultReprovisionBackoff = TimeSpan.FromSeconds(10);
/// <summary>
/// Cache key used when an artifact carries no cluster scoping (single-cluster or unscoped
/// deployments). A literal rather than an empty string so the row is visibly deliberate when
@@ -357,7 +365,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// push, <see cref="SpawnChild"/> and the supervision-reprovision handler all send from here, so a
/// child obtained by ANY route carries the same set. Cleared and repopulated whole on every apply,
/// alongside the routing maps, so renames/removals can never leave a stale entry. Empty until the
/// first apply — a child spawned before any apply is handed nothing, which is what it should have.</summary>
/// first apply — a child spawned before any apply is handed nothing, which is what it should have.
/// <para>It is only ever rebuilt by a SUCCESSFUL apply, so after a failed one it deliberately lingers on
/// the last-good composition — matching the address space, drivers and subscriptions that also keep
/// serving through a failed apply (#485/#486). A respawn or reprovision in that window therefore hands
/// the child the last-known-good set, not an empty one.</para></summary>
private readonly Dictionary<string, DesiredSubscriptionSet> _desiredByDriver = new(StringComparer.Ordinal);
/// <summary>
@@ -413,6 +425,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <see cref="DefaultApplyWedgeTimeout"/>).</summary>
private readonly TimeSpan _applyWedgeTimeout;
/// <summary>Per-driver rate limit for the supervision-reprovision hook (see
/// <see cref="DefaultReprovisionBackoff"/>).</summary>
private readonly TimeSpan _reprovisionBackoff;
/// <summary>When each driver was last reprovisioned after a supervision restart, so a crash loop is
/// paced rather than amplified. Bounded by the number of driver children; an entry for a driver that is
/// later removed simply stops being consulted (the handler resolves <c>_children</c> first).</summary>
private readonly Dictionary<string, DateTime> _lastReprovisionUtc = new(StringComparer.Ordinal);
/// <summary>
/// Host of this node's LocalDb replication partner — the only node that holds a copy of this
/// node's alarm queue, and therefore the only node it may ever stand down in favour of.
@@ -554,7 +575,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IDriverPrimaryRoleSource? primaryRoleSource = null,
IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null) =>
TimeSpan? applyWedgeTimeout = null,
TimeSpan? reprovisionBackoff = null) =>
// WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
// expression tree. Six IActorRef? parameters and several interface-typed ones mean a
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
@@ -566,7 +588,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider,
deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
sharesHistorizationScopeWithLocal, applyWedgeTimeout));
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff));
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
/// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -652,9 +674,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
IDriverPrimaryRoleSource? primaryRoleSource = null,
IHistorizationRoleView? historizationRoleView = null,
Func<CommonsNodeId, bool>? sharesHistorizationScopeWithLocal = null,
TimeSpan? applyWedgeTimeout = null)
TimeSpan? applyWedgeTimeout = null,
TimeSpan? reprovisionBackoff = null)
{
_applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
_reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff;
_historizationRoleView = historizationRoleView;
_sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal;
_telemetryHub = telemetryHub;
@@ -1089,6 +1113,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult);
Receive<RestartDriver>(HandleRestartDriver);
Receive<DriverInstanceActor.ReprovisionRequested>(HandleReprovisionRequested);
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
@@ -1145,6 +1170,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<DriverInstanceActor.ApplyResult>(HandleApplyResult);
Receive<RestartDriver>(HandleRestartDriver);
Receive<DriverInstanceActor.ReprovisionRequested>(HandleReprovisionRequested);
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
@@ -1728,6 +1754,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<RetryConfigDbConnection>(_ => TryRecoverFromStale());
Receive<RestartDriver>(HandleRestartDriver);
Receive<DriverInstanceActor.ReprovisionRequested>(HandleReprovisionRequested);
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
// The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
@@ -3012,7 +3039,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// </para>
/// </remarks>
/// <param name="msg">The restarted child's self-announcement.</param>
private void HandleReprovisionRequested(DriverInstanceActor.ReprovisionRequested msg)
private void HandleReprovisionRequested(DriverInstanceActor.ReprovisionRequested msg) =>
ReprovisionChild(msg, Sender);
/// <summary>A reprovision held back by <see cref="_reprovisionBackoff"/>, carrying the incarnation it
/// was requested by so the identity check still applies when it is re-delivered by the timer.</summary>
/// <param name="Request">The original request.</param>
/// <param name="Child">The incarnation that asked; re-validated against <c>_children</c> on delivery,
/// so a driver restarted or removed in the meantime discards it.</param>
private sealed record ReprovisionDeferred(
DriverInstanceActor.ReprovisionRequested Request, IActorRef Child);
private static string ReprovisionTimerKey(string driverInstanceId) => "reprovision-" + driverInstanceId;
private void ReprovisionChild(DriverInstanceActor.ReprovisionRequested msg, IActorRef child)
{
// Already stopped (a restart racing a reconcile that removed the driver) — nothing to provision.
if (!_children.TryGetValue(msg.DriverInstanceId, out var entry)) return;
@@ -3023,14 +3063,34 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// where InitializeRequested IS handled — it would start a second, concurrent InitializeAsync,
// racing the dispose-before-overwrite that #527's teardown exists to serialise
// (ModbusDriver._transport, MqttDriver._connection each leak a transport on that race).
if (!Equals(Sender, entry.Actor))
if (!Equals(child, entry.Actor))
{
_log.Debug(
"DriverHost {Node}: dropping a reprovision for {Id} from a superseded incarnation ({Stale})",
_localNode, msg.DriverInstanceId, Sender);
_localNode, msg.DriverInstanceId, child);
return;
}
// Rate-limit per driver (review F3). DEFERRED, never dropped: the last request must still be
// honoured, or a child whose final crash fell inside the window is left un-provisioned forever —
// exactly the 03/S16 failure this hook exists to close. One timer key per driver, so a burst
// collapses to a single pending reprovision rather than a queue of them.
var now = DateTime.UtcNow;
if (_lastReprovisionUtc.TryGetValue(msg.DriverInstanceId, out var last) &&
now - last < _reprovisionBackoff)
{
var delay = _reprovisionBackoff - (now - last);
_log.Warning(
"DriverHost {Node}: driver child {Id} was restarted by supervision again after {Reason} " +
"within {Backoff} — deferring the re-provision by {Delay} (crash loop?)",
_localNode, msg.DriverInstanceId, msg.Reason, _reprovisionBackoff, delay);
Timers.StartSingleTimer(
ReprovisionTimerKey(msg.DriverInstanceId), new ReprovisionDeferred(msg, child), delay);
return;
}
_lastReprovisionUtc[msg.DriverInstanceId] = now;
// A crash that Akka silently restarted is otherwise invisible to operators; name it.
_log.Warning(
"DriverHost {Node}: driver child {Id} was restarted by supervision after {Reason} — re-provisioning",
@@ -478,6 +478,115 @@ public sealed class DriverHostActorRestartTests : RuntimeActorTestBase
"concurrent InitializeAsync on a driver that was already connected");
}
/// <summary>
/// <b>Review finding F3 - a crash-looping driver must not become a reconnect storm.</b> There is no
/// <c>SupervisorStrategy</c> anywhere under <c>src/Server</c>, so the default decider restarts a
/// faulting child <b>immediately and without limit</b>. That was survivable while a restart did
/// nothing; the reprovision hook now makes every one of them do
/// <c>ShutdownAsync</c> + <c>InitializeAsync</c> + <c>SubscribeAsync</c>, so a driver that faults on
/// every value would hammer its device as fast as it can crash.
/// <para>
/// The reprovision is therefore rate-limited per driver. Crucially it is <b>deferred, not
/// dropped</b> - the last request must still be honoured, or a child whose final crash fell
/// inside the window would be left un-provisioned forever, which is the very failure this whole
/// plan exists to close. Two genuine back-to-back supervision restarts: the first reprovision
/// applies immediately, the second is held and then lands.
/// </para>
/// </summary>
[Fact]
public void A_second_reprovision_inside_the_backoff_window_is_deferred_not_dropped()
{
var db = NewInMemoryDbFactory();
// Generous relative to the in-memory step time below, so "inside the window" is a decision rather
// than a race, while still far under the presence budget the deferred landing waits on.
var backoff = TimeSpan.FromSeconds(3);
var factory = new RecordingSubscribableDriverFactory("Modbus");
var deploymentId = V3ArtifactSeeder.SeedV3Deployment(
db, RevA,
new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) },
Array.Empty<(string, string, string, string, string?)>(),
enabled: true);
V3ArtifactSeeder.SpawnHostAndApply(
this, db, TestNode, deploymentId, RevA, factory, reprovisionBackoff: backoff);
AwaitAssert(
() =>
{
factory.Created.Count.ShouldBe(1);
factory.Created[0].SubscribeCount.ShouldBe(1);
},
duration: PresenceBudget);
var driver = factory.Created[0];
// Crash 1: the first reprovision has no prior stamp, so it applies immediately.
driver.FireMalformedDataChange(SpeedPath);
AwaitAssert(
() =>
{
driver.InitializeCount.ShouldBe(2);
driver.SubscribeCount.ShouldBe(2, "the re-subscribe must land before the second crash is armed");
},
duration: PresenceBudget);
// Crash 2, immediately after - well inside the window.
driver.FireMalformedDataChange(SpeedPath);
AwaitAssert(
() => driver.ShutdownCount.ShouldBeGreaterThanOrEqualTo(
2, "the second crash seam did not trip supervision - nothing below is testing a backoff"),
duration: PresenceBudget);
driver.InitializeCount.ShouldBe(
2, "the second reprovision was applied immediately - the backoff is not rate-limiting anything");
// ...but it must still land once the window elapses.
AwaitAssert(
() => driver.InitializeCount.ShouldBe(
3, "the deferred reprovision was DROPPED, not deferred"),
duration: PresenceBudget);
}
/// <summary>
/// The <c>_children.Keys</c> arm of the desired-set authority's union. A driver with NO authored tags
/// is absent from both the value- and alarm-ref maps, so keying the authority off those maps alone
/// would leave it with no entry at all — and "absent" must not be allowed to differ from "empty",
/// because an empty set is what tells a child to <c>Unsubscribe</c> and drop a stale handle.
/// <para>
/// Asserted the only way it is observable: the tagless child is initialised (so it is provably
/// alive and was provisioned) and never subscribes, while its tag-owning sibling in the same
/// deployment does — the sibling is the positive control proving the apply's push pass actually
/// ran, so "it did not subscribe" cannot pass for "nothing happened".
/// </para>
/// </summary>
[Fact]
public void A_running_driver_with_no_authored_tags_is_handed_an_empty_set()
{
var db = NewInMemoryDbFactory();
var factory = new RecordingSubscribableDriverFactory("Modbus");
var deploymentId = V3ArtifactSeeder.SeedV3Deployment(
db, RevA,
new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) },
Array.Empty<(string, string, string, string, string?)>(),
enabled: true,
taglessDrivers: new[] { "DI-TAGLESS" });
V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory);
AwaitAssert(
() =>
{
factory.Created.Count.ShouldBe(2);
// Positive control: the sibling that DOES have a tag subscribed, so the push pass ran.
factory.Created.Single(d => d.DriverInstanceId == DriverId)
.SubscribeCount.ShouldBe(1);
factory.Created.Single(d => d.DriverInstanceId == "DI-TAGLESS")
.InitializeCount.ShouldBe(1);
},
duration: PresenceBudget);
factory.Created.Single(d => d.DriverInstanceId == "DI-TAGLESS")
.SubscribeCount.ShouldBe(
0, "a driver with no authored tags must be handed an EMPTY set, never a spurious subscribe");
}
/// <summary>
/// The verified negative, pinned so this plan's edits to the restart paths cannot break the one that
/// was already correct. Force Reconnect (<c>ReconnectDriver</c>) Tells <c>ForceReconnect</c> to the
@@ -47,18 +47,25 @@ public static class V3ArtifactSeeder
/// is parsed out of <c>TagConfig.alarm</c> by <c>TagConfigIntent</c>), which the alarm-ref half of the
/// restart suite needs — an alarm-bearing raw tag is excluded from the VALUE subscription set and carried
/// in the alarm one instead, so the two travel through different child code paths.</param>
/// <param name="taglessDrivers">Extra <c>DriverInstance</c> ids carrying NO devices and NO tags. The
/// only way to express a driver that is a running child yet absent from both the value- and alarm-ref
/// maps — the <c>_children.Keys</c> arm of <c>ApplyDesiredSubscriptions</c>'s union, which must still
/// hand that child an explicit EMPTY set rather than skipping it.</param>
/// <returns>The seeded deployment's id.</returns>
public static DeploymentId SeedV3Deployment(
IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev,
(string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] rawTags,
(string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs,
bool enabled = false,
IReadOnlyDictionary<string, string>? tagConfigByTag = null)
IReadOnlyDictionary<string, string>? tagConfigByTag = null,
string[]? taglessDrivers = null)
{
var drivers = rawTags
.Select(t => (t.Driver, t.DriverName))
.Distinct()
.Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled })
.Concat((taglessDrivers ?? Array.Empty<string>())
.Select(id => new { DriverInstanceId = id, RawFolderId = "rf-plant", Name = id, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = enabled }))
.ToArray();
var devices = rawTags
.Select(t => (t.Driver, t.Device))
@@ -130,14 +137,18 @@ public static class V3ArtifactSeeder
/// <param name="driverFactory">Optional driver factory forwarded to <c>DriverHostActor.Props</c>. Null (the
/// default, preserving every existing call site) leaves the host on its own default, which materialises no
/// real drivers. Supply one — with <c>enabled: true</c> on the seed — to actually spawn driver children.</param>
/// <param name="reprovisionBackoff">Optional per-driver supervision-reprovision rate limit forwarded to
/// <c>DriverHostActor.Props</c>; null keeps the production default. Exists so a test can observe the
/// deferral without waiting the production window.</param>
/// <returns>The host actor plus the publish and mux probes wired into it.</returns>
public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db,
NodeId localNode, DeploymentId deploymentId, RevisionHash rev,
IDriverFactory? driverFactory = null)
IDriverFactory? driverFactory = null,
TimeSpan? reprovisionBackoff = null)
{
var publish = kit.CreateTestProbe();
var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory);
var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff);
Dispatch(actor, coordinator, deploymentId, rev);
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(RuntimeActorTestBase.PresenceBudget);
@@ -157,14 +168,14 @@ public static class V3ArtifactSeeder
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db,
NodeId localNode, DeploymentId deploymentId, RevisionHash rev)
{
var (actor, mux, coordinator) = Spawn(kit, db, localNode, opcUaPublishActor: null, driverFactory: null);
var (actor, mux, coordinator) = Spawn(kit, db, localNode, opcUaPublishActor: null, driverFactory: null, reprovisionBackoff: null);
Dispatch(actor, coordinator, deploymentId, rev);
return (actor, mux);
}
private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db, NodeId localNode,
IActorRef? opcUaPublishActor, IDriverFactory? driverFactory)
IActorRef? opcUaPublishActor, IDriverFactory? driverFactory, TimeSpan? reprovisionBackoff)
{
var coordinator = kit.CreateTestProbe();
var mux = kit.CreateTestProbe();
@@ -177,7 +188,8 @@ public static class V3ArtifactSeeder
dependencyMux: mux.Ref,
opcUaPublishActor: opcUaPublishActor,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: vtHost.Ref));
virtualTagHostOverride: vtHost.Ref,
reprovisionBackoff: reprovisionBackoff));
return (actor, mux, coordinator);
}