fix(site-runtime): republish loaded static overrides to children/stream and answer pending waits on stop (review 2026-10-21 01/SR-U1, 01/SR-M3)
SR-U1: children are created before the async override load lands, so their trigger snapshots kept the deploy default forever after a restart/failover/ enable/sweep start. HandleOverridesLoaded now republishes every changed value through PublishAndNotifyChildren(evaluateWaiters:false), and skips attributes a static write touched before the load (newer than the stored row). SR-M3: PostStop cancels each WaitForAttribute timeout and replies 'instance stopped' instead of leaving scripts to time out and timeouts to dead-letter.
This commit is contained in:
@@ -132,6 +132,16 @@ public class InstanceActor : ReceiveActor
|
||||
|
||||
private Task _staticOverridePersistTail = Task.CompletedTask;
|
||||
|
||||
/// <summary>
|
||||
/// Review 2026-10-21 01/SR-U1: false until the PreStart static-override load has landed
|
||||
/// (successfully or not). While false, every static write records its attribute in
|
||||
/// <see cref="_staticWritesBeforeOverrideLoad"/> so the load cannot roll a newer in-memory
|
||||
/// value back to the older stored one — a race that only mattered once the load started
|
||||
/// republishing what it applies.
|
||||
/// </summary>
|
||||
private bool _overridesLoaded;
|
||||
private readonly HashSet<string> _staticWritesBeforeOverrideLoad = new();
|
||||
|
||||
internal sealed record StaticOverridesFlushed(IActorRef Actor, Exception? Error);
|
||||
|
||||
/// <summary>
|
||||
@@ -439,6 +449,20 @@ public class InstanceActor : ReceiveActor
|
||||
state.Timer?.Cancel();
|
||||
_tagSubscribeStates.Clear();
|
||||
|
||||
// Review 2026-10-21 01/SR-M3: answer every pending WaitForAttribute now. Left alone, each
|
||||
// waiter's scheduled WaitForAttributeTimeout dead-letters later (dead letters are a
|
||||
// monitored site-health metric) and the waiting script burns its full wait timeout plus
|
||||
// the Ask timeout, then reports "timed out" — indistinguishable from a device that never
|
||||
// answered, when the truth is that the instance went away (redeploy/disable/delete).
|
||||
foreach (var (correlationId, pending) in _attributeWaiters)
|
||||
{
|
||||
pending.Timeout.Cancel();
|
||||
pending.Replyer.Tell(new WaitForAttributeResponse(
|
||||
correlationId, Matched: false, null, null, TimedOut: false,
|
||||
ErrorMessage: $"Wait for '{pending.AttributeName}' abandoned: instance stopped"));
|
||||
}
|
||||
_attributeWaiters.Clear();
|
||||
|
||||
// SQLite's async APIs may block synchronously. Persist on the ordered
|
||||
// thread-pool chain and acknowledge completion separately from Terminated.
|
||||
_staticOverrideFlushTimer?.Cancel();
|
||||
@@ -606,6 +630,9 @@ public class InstanceActor : ReceiveActor
|
||||
_attributes[command.AttributeName] = command.Value;
|
||||
}
|
||||
|
||||
if (!_overridesLoaded)
|
||||
_staticWritesBeforeOverrideLoad.Add(command.AttributeName);
|
||||
|
||||
// Publish attribute change to stream and notify children
|
||||
var changed = new AttributeValueChanged(
|
||||
_instanceUniqueName,
|
||||
@@ -1798,10 +1825,34 @@ public class InstanceActor : ReceiveActor
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Applies static overrides loaded from SQLite on top of default values.
|
||||
/// Applies static overrides loaded from SQLite on top of default values, then republishes
|
||||
/// every value that actually changed (review 2026-10-21 01/SR-U1).
|
||||
///
|
||||
/// <para>
|
||||
/// The republish is the point of the fix. <see cref="CreateChildActors"/> runs synchronously
|
||||
/// in <see cref="PreStart"/>, BEFORE this load result arrives, so every Script/Alarm Actor's
|
||||
/// trigger snapshot was seeded with the deploy DEFAULT; and a static attribute never re-emits
|
||||
/// a change on its own, so after a restart, failover, enable or sweep start the children
|
||||
/// evaluated against the default for the actor's whole lifetime while <c>GetAttribute</c>
|
||||
/// answered with the override. Routing each changed value through
|
||||
/// <see cref="PublishAndNotifyChildren"/> re-seeds every child snapshot and the site stream in
|
||||
/// one place. Consequence, deliberately accepted: a ValueChange/Conditional script keyed on an
|
||||
/// overridden static attribute fires once as the actor starts — exactly as one keyed on a
|
||||
/// data-sourced attribute already fires on that attribute's first DCL value.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Waiters are NOT evaluated: none can be waiting on a value the instance already held, and a
|
||||
/// replay must never satisfy a <c>WaitAsync</c> registered against the live value. An
|
||||
/// attribute a static write touched before this load landed is skipped — the write is newer
|
||||
/// than the stored row and stays authoritative.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
private void HandleOverridesLoaded(LoadOverridesResult result)
|
||||
{
|
||||
_overridesLoaded = true;
|
||||
var writtenBeforeLoad = new HashSet<string>(_staticWritesBeforeOverrideLoad);
|
||||
_staticWritesBeforeOverrideLoad.Clear();
|
||||
|
||||
if (result.Error != null)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
@@ -1810,8 +1861,15 @@ public class InstanceActor : ReceiveActor
|
||||
return;
|
||||
}
|
||||
|
||||
var changedAttributes = new List<string>();
|
||||
foreach (var kvp in result.Overrides)
|
||||
{
|
||||
if (writtenBeforeLoad.Contains(kvp.Key))
|
||||
continue;
|
||||
|
||||
_attributes.TryGetValue(kvp.Key, out var previousValue);
|
||||
_attributeQualities.TryGetValue(kvp.Key, out var previousQuality);
|
||||
|
||||
// Persisted override values are canonical strings — a JSON array
|
||||
// string for List attributes, a plain string for scalars. Decode List
|
||||
// overrides to a typed list (matching the config-default load), set
|
||||
@@ -1852,11 +1910,31 @@ public class InstanceActor : ReceiveActor
|
||||
{
|
||||
_attributes[kvp.Key] = kvp.Value;
|
||||
}
|
||||
|
||||
// Only attributes the deployed config defines are republished: an orphan override row
|
||||
// (an attribute since removed from the template) has no child interested in it.
|
||||
if (_resolvedAttributeByName.ContainsKey(kvp.Key) &&
|
||||
(!string.Equals(AttributeValueCodec.Encode(previousValue),
|
||||
AttributeValueCodec.Encode(_attributes[kvp.Key]), StringComparison.Ordinal) ||
|
||||
!string.Equals(previousQuality, _attributeQualities.GetValueOrDefault(kvp.Key),
|
||||
StringComparison.Ordinal)))
|
||||
{
|
||||
changedAttributes.Add(kvp.Key);
|
||||
}
|
||||
}
|
||||
|
||||
foreach (var name in changedAttributes)
|
||||
{
|
||||
PublishAndNotifyChildren(
|
||||
new AttributeValueChanged(
|
||||
_instanceUniqueName, name, name, _attributes[name],
|
||||
_attributeQualities.GetValueOrDefault(name) ?? "Good", DateTimeOffset.UtcNow),
|
||||
evaluateWaiters: false);
|
||||
}
|
||||
|
||||
_logger.LogDebug(
|
||||
"Loaded {Count} static overrides for {Instance}",
|
||||
result.Overrides.Count, _instanceUniqueName);
|
||||
"Loaded {Count} static overrides for {Instance} ({Changed} republished)",
|
||||
result.Overrides.Count, _instanceUniqueName, changedAttributes.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user