feat(site-runtime): ConnectionQualityChanged(Good) resets tag-subscribe backoff to the floor (Lever 1); lock in superseded-reply drop
Task 4 review fix-ups: RepeatedFailures gaps[1] lower bound 600→500 (two compounding jitter draws bottom out at 512 ms); SubscribeToDcl also clears _tagPathToAttributes and its prune is documented as defensive (PreStart is the only caller).
This commit is contained in:
@@ -186,8 +186,8 @@ public class InstanceActor : ReceiveActor
|
|||||||
// (a tick while a young request is outstanding re-arms instead of re-sending) and
|
// (a tick while a young request is outstanding re-arms instead of re-sending) and
|
||||||
// exponential backoff (see NextTagSubscribeInterval) so a slow gateway is never
|
// exponential backoff (see NextTagSubscribeInterval) so a slow gateway is never
|
||||||
// handed a fresh N-tag subscribe on top of the one it is still processing.
|
// handed a fresh N-tag subscribe on top of the one it is still processing.
|
||||||
// A re-entry into SubscribeToDcl rebuilds all three together: the retry state is
|
// SubscribeToDcl has a single caller today (PreStart); the prune at its top is defensive, so a
|
||||||
// pruned (timers cancelled, pending rows dropped) before the new grouping is built.
|
// future second call site rebuilds the retry state and both tag maps together instead of stranding them.
|
||||||
private readonly Dictionary<string, List<string>> _tagsByConnection = new();
|
private readonly Dictionary<string, List<string>> _tagsByConnection = new();
|
||||||
private readonly Dictionary<string, string> _pendingTagSubscribes = new();
|
private readonly Dictionary<string, string> _pendingTagSubscribes = new();
|
||||||
private readonly Dictionary<string, TagSubscribeState> _tagSubscribeStates = new();
|
private readonly Dictionary<string, TagSubscribeState> _tagSubscribeStates = new();
|
||||||
@@ -1171,6 +1171,20 @@ public class InstanceActor : ReceiveActor
|
|||||||
_logger.LogWarning("Connection {Connection} quality changed to {Quality} for instance {Instance}",
|
_logger.LogWarning("Connection {Connection} quality changed to {Quality} for instance {Instance}",
|
||||||
qualityChanged.ConnectionName, qualityChanged.Quality, _instanceUniqueName);
|
qualityChanged.ConnectionName, qualityChanged.Quality, _instanceUniqueName);
|
||||||
|
|
||||||
|
// Lever 1 (2026-08-28): a connection returning to Good is the Instance Actor's
|
||||||
|
// only restore signal (the DCL re-subscribes transparently). If our own subscribe
|
||||||
|
// on it is still unconfirmed, retry at the floor now rather than in up to
|
||||||
|
// TagSubscribeRetryMaxIntervalMs — recovery must be prompt. Timer != null ⇔ the
|
||||||
|
// subscribe is unconfirmed (it is only nulled by a Success); if a request is in
|
||||||
|
// flight and young the re-armed tick just hits the guard and re-arms at the floor.
|
||||||
|
if (qualityChanged.Quality == QualityCode.Good
|
||||||
|
&& _tagSubscribeStates.TryGetValue(qualityChanged.ConnectionName, out var subState)
|
||||||
|
&& subState.Timer != null)
|
||||||
|
{
|
||||||
|
ResetTagSubscribeBackoff(subState);
|
||||||
|
ArmTagSubscribeWatchdog(qualityChanged.ConnectionName, subState);
|
||||||
|
}
|
||||||
|
|
||||||
if (_configuration == null) return;
|
if (_configuration == null) return;
|
||||||
|
|
||||||
// Mark all attributes bound to this connection with the new quality
|
// Mark all attributes bound to this connection with the new quality
|
||||||
@@ -1208,6 +1222,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
{
|
{
|
||||||
if (_dclManager == null || _configuration == null) return;
|
if (_dclManager == null || _configuration == null) return;
|
||||||
|
|
||||||
|
// Defensive — PreStart is the only caller today.
|
||||||
// Lever 1: prune retry state from a previous build — cancels any armed watchdog and
|
// Lever 1: prune retry state from a previous build — cancels any armed watchdog and
|
||||||
// drops its pending row so a re-entry can never strand an in-flight slot.
|
// drops its pending row so a re-entry can never strand an in-flight slot.
|
||||||
foreach (var state in _tagSubscribeStates.Values)
|
foreach (var state in _tagSubscribeStates.Values)
|
||||||
@@ -1221,6 +1236,7 @@ public class InstanceActor : ReceiveActor
|
|||||||
// Group attributes by their bound connection name into the persistent
|
// Group attributes by their bound connection name into the persistent
|
||||||
// _tagsByConnection field so a retry can re-send the exact same request.
|
// _tagsByConnection field so a retry can re-send the exact same request.
|
||||||
_tagsByConnection.Clear();
|
_tagsByConnection.Clear();
|
||||||
|
_tagPathToAttributes.Clear();
|
||||||
foreach (var attr in _configuration.Attributes)
|
foreach (var attr in _configuration.Attributes)
|
||||||
{
|
{
|
||||||
if (string.IsNullOrEmpty(attr.DataSourceReference) ||
|
if (string.IsNullOrEmpty(attr.DataSourceReference) ||
|
||||||
|
|||||||
@@ -1410,7 +1410,8 @@ public class InstanceActorTests : TestKit, IDisposable
|
|||||||
|
|
||||||
// Expected nominal gaps: 400, 800(cap), 800(cap) — each ±20 % jitter and scheduler slack.
|
// Expected nominal gaps: 400, 800(cap), 800(cap) — each ±20 % jitter and scheduler slack.
|
||||||
Assert.InRange(gaps[0].TotalMilliseconds, 300, 560);
|
Assert.InRange(gaps[0].TotalMilliseconds, 300, 560);
|
||||||
Assert.InRange(gaps[1].TotalMilliseconds, 600, 1100);
|
// gaps[1] compounds two jitter draws: 2×(2×200×0.8)×0.8 = 512 ms minimum.
|
||||||
|
Assert.InRange(gaps[1].TotalMilliseconds, 500, 1100);
|
||||||
Assert.InRange(gaps[2].TotalMilliseconds, 600, 1100);
|
Assert.InRange(gaps[2].TotalMilliseconds, 600, 1100);
|
||||||
|
|
||||||
// Success resets: no further request follows a confirmed subscription.
|
// Success resets: no further request follows a confirmed subscription.
|
||||||
@@ -1438,4 +1439,65 @@ public class InstanceActorTests : TestKit, IDisposable
|
|||||||
var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||||||
Assert.NotEqual(first.CorrelationId, retry.CorrelationId);
|
Assert.NotEqual(first.CorrelationId, retry.CorrelationId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- Lever 1 (2026-08-28): Good quality resets the backoff ---
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void ConnectionBackToGood_ResetsBackoff_AndRetriesPromptly()
|
||||||
|
{
|
||||||
|
var options = new SiteRuntimeOptions
|
||||||
|
{
|
||||||
|
TagSubscribeRetryIntervalMs = 200,
|
||||||
|
TagSubscribeRetryMaxIntervalMs = 5000,
|
||||||
|
TagSubscribeInFlightCeilingMs = 60_000,
|
||||||
|
};
|
||||||
|
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||||
|
|
||||||
|
// Drive the interval up: 3 failures → 400, 800, 1600 (nominal).
|
||||||
|
var req = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||||
|
for (var i = 0; i < 3; i++)
|
||||||
|
{
|
||||||
|
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||||||
|
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
req = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||||||
|
}
|
||||||
|
// One more failure would nominally wait ~3.2 s. Instead the connection recovers:
|
||||||
|
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||||||
|
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
instance.Tell(new ConnectionQualityChanged("conn-1", QualityCode.Good, DateTimeOffset.UtcNow));
|
||||||
|
|
||||||
|
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||||
|
dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(2));
|
||||||
|
Assert.True(sw.Elapsed < TimeSpan.FromMilliseconds(1000), $"retry took {sw.Elapsed} — backoff not reset");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void LateReplyForFailedAndReissuedRequest_IsIgnored()
|
||||||
|
{
|
||||||
|
var options = new SiteRuntimeOptions
|
||||||
|
{
|
||||||
|
TagSubscribeRetryIntervalMs = 200,
|
||||||
|
TagSubscribeRetryMaxIntervalMs = 200,
|
||||||
|
// Short ceiling on purpose: the second request stays in flight (nothing ever
|
||||||
|
// answers it), so the in-flight guard is what re-sends. With a 60 s ceiling the
|
||||||
|
// guard would hold for 60 s and the test could not tell "late reply ignored"
|
||||||
|
// apart from "late reply wrongly cleared the slot" — both look like silence.
|
||||||
|
TagSubscribeInFlightCeilingMs = 500,
|
||||||
|
};
|
||||||
|
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||||
|
|
||||||
|
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||||
|
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||||||
|
false, "busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||||||
|
|
||||||
|
// Duplicate/late Success for the FIRST id: already removed from the pending map → ignored,
|
||||||
|
// so the watchdog guarding the SECOND is untouched and still re-sends once its
|
||||||
|
// in-flight ceiling expires. Had the late reply been honoured it would have cancelled
|
||||||
|
// that watchdog and cleared the slot, and no third request would ever be sent.
|
||||||
|
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||||||
|
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||||||
|
Assert.NotEqual(second.CorrelationId, third.CorrelationId);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user