fix(runtime): drop value bursts from driver children the host already stopped
Review I-1. Context.Stop is asynchronous, so a removed child kept publishing through the rest of the apply handler and its own stop. Those bursts were forwarded after SourceRemoved and re-retained the removed driver's last Good in the mux, where nothing would ever clear it while a replay subscriber held the ref (a removed driver never publishes again, and its PostStop Bad fan resolves no refs once the index is rebuilt without it). StopChild and HandleRestartDriver now retire the child (_retiredChildren + Context.Watch) before stopping it. ForwardBatchToMux drops a burst whose Sender is retired, and Terminated (handled in Steady, Applying and Stale) removes the entry. Akka delivers Terminated after the watched child's last message, so the set is bounded and a live incarnation is never filtered. ConnectivityChanged from a retired child is deliberately not filtered, because the respawn path relies on its Bad fan. Tests (review m-3 folded in): a stale burst spoofed as the removed child, held alive in PostStop by a new RecordingSubscribableDriver.ShutdownGate, no longer re-retains (red with the guard disabled). New cases cover a config-change respawn and an operator Restart, each with the new driver's value replaying as the positive control. The arrange step now waits for the real subscription set and fishes for both drivers' values at once (two sequential fishes could discard the other burst). Refs #731 #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -205,6 +205,17 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
|
||||
/// <summary>Node-wide script run-guard options for the scripted-alarm engine; null ⇒ compiled defaults.</summary>
|
||||
private readonly ScriptingOptions? _scripting;
|
||||
|
||||
/// <summary>
|
||||
/// Driver children this host has stopped (<see cref="StopChild"/>, an operator Restart) and is watching
|
||||
/// until they terminate. A value burst from one of them is DROPPED in <see cref="ForwardBatchToMux"/>:
|
||||
/// <c>Context.Stop</c> is asynchronous, so the old child keeps publishing for the rest of the apply
|
||||
/// handler and its own stop, and forwarding those bursts after <see cref="EvictRetainedValues"/> would
|
||||
/// re-retain a removed driver's last Good in the dependency mux indefinitely. Akka delivers
|
||||
/// <see cref="Terminated"/> after the last message the watched child sent here, so removing the entry
|
||||
/// there bounds the set without ever dropping a live incarnation's values.
|
||||
/// </summary>
|
||||
private readonly HashSet<IActorRef> _retiredChildren = new();
|
||||
private readonly IActorRef? _scriptedAlarmHostOverride;
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
|
||||
@@ -1516,6 +1527,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
Receive<ApplyWedged>(HandleApplyWedged);
|
||||
Receive<GetDiagnostics>(HandleGetDiagnostics);
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);
|
||||
Receive<Terminated>(OnRetiredChildTerminated);
|
||||
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
|
||||
Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged);
|
||||
Receive<CommsRecoveryAudit>(HandleCommsRecoveryAudit);
|
||||
@@ -1578,6 +1590,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
Receive<ApplyWedged>(HandleApplyWedged);
|
||||
Receive<GetDiagnostics>(HandleGetDiagnostics);
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);
|
||||
Receive<Terminated>(OnRetiredChildTerminated);
|
||||
Receive<DriverInstanceActor.AttributeAlarmPublished>(ForwardNativeAlarm);
|
||||
Receive<DriverInstanceActor.ConnectivityChanged>(OnDriverConnectivityChanged);
|
||||
Receive<CommsRecoveryAudit>(HandleCommsRecoveryAudit);
|
||||
@@ -1622,6 +1635,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// </summary>
|
||||
private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg)
|
||||
{
|
||||
// A burst from an incarnation this host already stopped: forwarding it would undo the SourceRemoved
|
||||
// eviction (see _retiredChildren). Its ConnectivityChanged is NOT filtered — the respawn path relies on
|
||||
// that Bad fan to cover the gap.
|
||||
if (_retiredChildren.Contains(Sender))
|
||||
{
|
||||
_log.Debug("DriverHost {Node}: dropped a {Count}-item value burst from a stopped incarnation of {Id}",
|
||||
_localNode, msg.Items.Count, msg.DriverInstanceId);
|
||||
return;
|
||||
}
|
||||
|
||||
// One envelope per burst; the mux filters per item (it registered a batch receive).
|
||||
_dependencyMux?.Tell(msg);
|
||||
|
||||
@@ -2517,6 +2540,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(m =>
|
||||
_log.Debug("DriverHost {Node}: dropping a {Count}-item value burst from {Id} while Stale (no address space)",
|
||||
_localNode, m.Items.Count, m.DriverInstanceId));
|
||||
Receive<Terminated>(OnRetiredChildTerminated);
|
||||
Receive<DriverInstanceActor.AttributeAlarmPublished>(m =>
|
||||
_log.Debug("DriverHost {Node}: dropping a native alarm transition ({Cond}) from {Id} while Stale",
|
||||
_localNode, m.Args.ConditionId, m.DriverInstanceId));
|
||||
@@ -3832,6 +3856,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
{
|
||||
if (!_children.TryGetValue(driverInstanceId, out var entry)) return;
|
||||
StopDependencyAdapter(driverInstanceId);
|
||||
RetireChild(entry.Actor);
|
||||
Context.Stop(entry.Actor);
|
||||
_children.Remove(driverInstanceId);
|
||||
EvictRetainedValues(driverInstanceId);
|
||||
@@ -3863,15 +3888,28 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// values travel, so the eviction is ordered ahead of them.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Residual: a burst the old child enqueued on this host just before it stopped can still be forwarded
|
||||
/// after the eviction and re-retain a value. The window is one mailbox drain; a comms-loss fan or the
|
||||
/// next real publish overwrites it, and the mux's grace sweep bounds it once no replay subscriber holds
|
||||
/// the ref.
|
||||
/// The stopped child keeps publishing until it actually terminates, and for a REMOVED driver nothing
|
||||
/// would ever overwrite a value re-retained after this eviction (it never publishes again, and its
|
||||
/// PostStop Bad fan resolves no refs once the apply has rebuilt the index without it). So the caller
|
||||
/// also <see cref="RetireChild"/>s it, and <see cref="ForwardBatchToMux"/> drops every burst from a
|
||||
/// retired child.
|
||||
/// </remarks>
|
||||
/// <param name="driverInstanceId">The driver instance that ended.</param>
|
||||
private void EvictRetainedValues(string driverInstanceId) =>
|
||||
_dependencyMux?.Tell(new DependencyMuxActor.SourceRemoved(driverInstanceId));
|
||||
|
||||
/// <summary>Marks a child this host is about to stop as retired and watches it until it terminates.</summary>
|
||||
/// <param name="child">The child being stopped.</param>
|
||||
private void RetireChild(IActorRef child)
|
||||
{
|
||||
_retiredChildren.Add(child);
|
||||
Context.Watch(child);
|
||||
}
|
||||
|
||||
/// <summary>A retired child has terminated; nothing more can arrive from it.</summary>
|
||||
/// <param name="msg">The termination notice.</param>
|
||||
private void OnRetiredChildTerminated(Terminated msg) => _retiredChildren.Remove(msg.ActorRef);
|
||||
|
||||
private static string ActorNameFor(string driverInstanceId, long generation)
|
||||
{
|
||||
// Akka actor names cannot contain '/', ':', or whitespace. Mangle defensively. The monotonic
|
||||
@@ -3917,6 +3955,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
|
||||
// Stop the existing child actor — DriverInstanceActor.PostStop calls ShutdownAsync.
|
||||
StopDependencyAdapter(msg.DriverInstanceId);
|
||||
RetireChild(entry.Actor);
|
||||
Context.Stop(entry.Actor);
|
||||
_children.Remove(msg.DriverInstanceId);
|
||||
// An operator Restart builds a NEW driver object, so the old one's retained values are evicted like a
|
||||
|
||||
Reference in New Issue
Block a user