fix(runtime): malformed artifact bytes on boot/restore are no answer (#601); evict telemetry snapshots on driver removal (#609)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
using System.Diagnostics;
|
||||
using System.Text.Json;
|
||||
using Akka.Actor;
|
||||
using Akka.Cluster.Tools.PublishSubscribe;
|
||||
using Akka.Event;
|
||||
@@ -1097,7 +1098,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
{
|
||||
var correlation = CorrelationId.NewId();
|
||||
|
||||
ReconcileDriversFromBlob(deploymentId, blob);
|
||||
// #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration.
|
||||
// Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
|
||||
// #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration.
|
||||
// Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
|
||||
if (ReconcileDriversFromBlob(deploymentId, blob) is null)
|
||||
return;
|
||||
|
||||
// Hand the cached blob to the rebuild so it materialises the client-facing address space from
|
||||
// it directly. Passing only the deploymentId would drive a ConfigDb read — unreachable here by
|
||||
@@ -2483,7 +2489,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
return null;
|
||||
}
|
||||
|
||||
ApplyDriverSpecs(DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value));
|
||||
// #601 (S-10): non-empty but malformed bytes are "no answer" too. The lenient static
|
||||
// ParseDriverInstances swallows the JsonException into an empty list, and an empty list plans every
|
||||
// running child for StopChild. Open throws instead, so a corrupt cache row / restored blob leaves
|
||||
// the running drivers alone, exactly like the zero-length case above.
|
||||
IReadOnlyList<DriverInstanceSpec> specs;
|
||||
try
|
||||
{
|
||||
using var parsed = DeploymentArtifact.Open(blob);
|
||||
specs = parsed.ParseDriverInstances(_localNode.Value);
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_log.Error(ex,
|
||||
"DriverHost {Node}: artifact for {Id} is malformed ({Bytes} bytes); skipping reconcile and keeping the {Count} running driver(s)",
|
||||
_localNode, deploymentId, blob.Length, _children.Count);
|
||||
return null;
|
||||
}
|
||||
|
||||
ApplyDriverSpecs(specs);
|
||||
return blob;
|
||||
}
|
||||
|
||||
@@ -2772,10 +2796,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
return;
|
||||
}
|
||||
|
||||
// #601 (S-10): Open, not the lenient static ParseComposition — that wrapper swallows malformed JSON
|
||||
// into an EMPTY composition, which would hand every child an empty desired set and drop its live
|
||||
// subscription. A malformed artifact is no answer: leave the subscriptions as they are.
|
||||
AddressSpaceComposition composition;
|
||||
try
|
||||
{
|
||||
composition = DeploymentArtifact.ParseComposition(blob, _localNode.Value);
|
||||
using var parsed = DeploymentArtifact.Open(blob);
|
||||
composition = parsed.ParseComposition(_localNode.Value);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
@@ -3204,6 +3232,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
StopDependencyAdapter(driverInstanceId);
|
||||
Context.Stop(entry.Actor);
|
||||
_children.Remove(driverInstanceId);
|
||||
// #609: the instance left the plan (StopChild is only reached via plan.ToStop), so its cached
|
||||
// Health/Resilience snapshots must not replay to new telemetry subscribers.
|
||||
_telemetryHub?.Evict(driverInstanceId);
|
||||
_log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId);
|
||||
}
|
||||
|
||||
|
||||
@@ -84,4 +84,11 @@ public interface ITelemetryLocalHub
|
||||
/// instances (a handful), so a normal large capacity leaves ample headroom.
|
||||
/// </param>
|
||||
ITelemetrySubscription Subscribe(int boundedCapacity);
|
||||
|
||||
/// <summary>
|
||||
/// Drops the cached Health and Resilience snapshots of a driver instance that has been removed from
|
||||
/// this node, so a later <see cref="Subscribe"/> no longer replays its last-known state. Live
|
||||
/// subscribers are not told; the instance simply stops emitting. Unknown ids are a no-op.
|
||||
/// </summary>
|
||||
void Evict(string driverInstanceId);
|
||||
}
|
||||
|
||||
@@ -37,9 +37,8 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub
|
||||
|
||||
// Snapshot last-value caches. Only ever mutated under _gate (kept ConcurrentDictionary so a
|
||||
// Subscribe reading them under the gate is trivially safe even against any future lock-free read).
|
||||
// TODO(mesh-phase5+): evict snapshot cache entries on driver-instance removal (needs a
|
||||
// driver-lifecycle hook); pre-production, accepted for now. Until then a decommissioned driver
|
||||
// instance's last-known Health/Resilience state replays to new subscribers forever.
|
||||
// Entries are evicted via Evict when DriverHostActor removes the instance (#609), so a
|
||||
// decommissioned driver's last-known state does not replay to new subscribers.
|
||||
private readonly ConcurrentDictionary<string, TelemetryItem.Health> _healthCache = new();
|
||||
private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new();
|
||||
|
||||
@@ -66,6 +65,20 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Evict(string driverInstanceId)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(driverInstanceId);
|
||||
|
||||
lock (_gate)
|
||||
{
|
||||
_healthCache.TryRemove(driverInstanceId, out _);
|
||||
foreach (var key in _resilienceCache.Keys)
|
||||
if (string.Equals(key.InstanceId, driverInstanceId, StringComparison.Ordinal))
|
||||
_resilienceCache.TryRemove(key, out _);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ITelemetrySubscription Subscribe(int boundedCapacity)
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user