feat(site-runtime): tag-subscribe failure reply frees the in-flight slot, backs off (×2 ±20%, capped) and Success resets to floor (Lever 1)
Also from the Task 3 review: FailedSubscribeResponse_IsRetried guards the failure path again (large ceiling); HandleSubscribeTagsResponse reads state instead of get-or-create; SubscribeToDcl prunes stale retry state on re-entry; doc comments refreshed. Claude-Session: https://claude.ai/code/session_019ZxXT5sxZs5XrtdwYzJDPV
This commit is contained in:
@@ -186,6 +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
|
||||||
|
// pruned (timers cancelled, pending rows dropped) before the new grouping is built.
|
||||||
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();
|
||||||
@@ -1206,6 +1208,16 @@ public class InstanceActor : ReceiveActor
|
|||||||
{
|
{
|
||||||
if (_dclManager == null || _configuration == null) return;
|
if (_dclManager == null || _configuration == null) return;
|
||||||
|
|
||||||
|
// 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.
|
||||||
|
foreach (var state in _tagSubscribeStates.Values)
|
||||||
|
{
|
||||||
|
state.Timer?.Cancel();
|
||||||
|
if (state.InFlightCorrelationId != null)
|
||||||
|
_pendingTagSubscribes.Remove(state.InFlightCorrelationId);
|
||||||
|
}
|
||||||
|
_tagSubscribeStates.Clear();
|
||||||
|
|
||||||
// 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();
|
||||||
@@ -1348,18 +1360,27 @@ public class InstanceActor : ReceiveActor
|
|||||||
state.Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs);
|
state.Interval = TimeSpan.FromMilliseconds(_options.TagSubscribeRetryIntervalMs);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Handles the DCL's tag-subscribe response: on Success cancels the connection's
|
/// Handles the DCL's tag-subscribe response. A reply whose correlation id is not the
|
||||||
/// retry timer; on failure logs + emits a site event and leaves the armed timer
|
/// pending one (superseded at the in-flight ceiling, or a duplicate) is ignored. On
|
||||||
/// to re-send. A reply for an unknown/superseded correlation id is ignored.
|
/// Success the connection's watchdog is cancelled, the in-flight slot cleared and the
|
||||||
|
/// backoff reset to the floor. On failure (Lever 1, 2026-08-28) the slot is freed, the
|
||||||
|
/// backoff advanced (×2 ±20 % jitter, capped) and the watchdog re-armed at the new
|
||||||
|
/// interval, so the next attempt waits the backed-off interval — not the in-flight
|
||||||
|
/// ceiling — and duplicates are still impossible.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private void HandleSubscribeTagsResponse(SubscribeTagsResponse response)
|
private void HandleSubscribeTagsResponse(SubscribeTagsResponse response)
|
||||||
{
|
{
|
||||||
if (!_pendingTagSubscribes.Remove(response.CorrelationId, out var connectionName))
|
if (!_pendingTagSubscribes.Remove(response.CorrelationId, out var connectionName))
|
||||||
return; // stale/superseded reply — ignore
|
return; // stale/superseded reply — ignore
|
||||||
|
|
||||||
|
// Read, never create: the id was minted by SendTagSubscribe, which creates the
|
||||||
|
// state, so a missing entry means the retry state was pruned (SubscribeToDcl
|
||||||
|
// re-entry) and this reply no longer governs anything.
|
||||||
|
if (!_tagSubscribeStates.TryGetValue(connectionName, out var state))
|
||||||
|
return;
|
||||||
|
|
||||||
if (response.Success)
|
if (response.Success)
|
||||||
{
|
{
|
||||||
var state = GetTagSubscribeState(connectionName);
|
|
||||||
if (state.InFlightCorrelationId == response.CorrelationId)
|
if (state.InFlightCorrelationId == response.CorrelationId)
|
||||||
{
|
{
|
||||||
state.Timer?.Cancel();
|
state.Timer?.Cancel();
|
||||||
@@ -1373,14 +1394,21 @@ public class InstanceActor : ReceiveActor
|
|||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
|
if (state.InFlightCorrelationId == response.CorrelationId)
|
||||||
|
{
|
||||||
|
// A definite failure: the slot is free, back off, and re-arm at the NEW
|
||||||
|
// interval so the backoff is honoured on this very retry, not the one after.
|
||||||
|
state.InFlightCorrelationId = null;
|
||||||
|
BackOffTagSubscribe(state);
|
||||||
|
ArmTagSubscribeWatchdog(connectionName, state);
|
||||||
|
}
|
||||||
_logger.LogWarning(
|
_logger.LogWarning(
|
||||||
"Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry",
|
"Instance {Instance} tag subscribe FAILED on connection {Connection}: {Error} — will retry in {Interval}",
|
||||||
_instanceUniqueName, connectionName, response.ErrorMessage);
|
_instanceUniqueName, connectionName, response.ErrorMessage, state.Interval);
|
||||||
_ = _serviceProvider?.GetService<ISiteEventLogger>()?.LogEventAsync(
|
_ = _serviceProvider?.GetService<ISiteEventLogger>()?.LogEventAsync(
|
||||||
"instance_lifecycle", "Warning", _instanceUniqueName,
|
"instance_lifecycle", "Warning", _instanceUniqueName,
|
||||||
$"InstanceActor:{_instanceUniqueName}",
|
$"InstanceActor:{_instanceUniqueName}",
|
||||||
$"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry");
|
$"Tag subscribe failed on connection '{connectionName}': {response.ErrorMessage} — will retry in {state.Interval}");
|
||||||
// Leave the armed retry timer to re-send.
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1290,7 +1290,7 @@ public class InstanceActorTests : TestKit, IDisposable
|
|||||||
[Fact]
|
[Fact]
|
||||||
public void FailedSubscribeResponse_IsRetried()
|
public void FailedSubscribeResponse_IsRetried()
|
||||||
{
|
{
|
||||||
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 200 };
|
var options = new SiteRuntimeOptions { TagSubscribeRetryIntervalMs = 200, TagSubscribeInFlightCeilingMs = 60_000 };
|
||||||
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||||
|
|
||||||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||||
@@ -1370,7 +1370,8 @@ public class InstanceActorTests : TestKit, IDisposable
|
|||||||
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||||
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
var second = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||||||
|
|
||||||
// A late Success for the FIRST id must not cancel the watchdog guarding the SECOND.
|
// A late Success for the FIRST id is dropped at the pending-map check (its id was
|
||||||
|
// removed at the ceiling); the watchdog guarding the SECOND is untouched.
|
||||||
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||||||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // second eventually re-sent
|
var third = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3)); // second eventually re-sent
|
||||||
@@ -1381,4 +1382,60 @@ public class InstanceActorTests : TestKit, IDisposable
|
|||||||
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800));
|
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(800));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- Lever 1 (2026-08-28): backoff on failure, reset on success ---
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void RepeatedFailures_BackOff_AndSuccessResetsToFloor()
|
||||||
|
{
|
||||||
|
// Floor 200 ms, max 800 ms. Jitter is ±20 % so bound the assertions, don't pin them.
|
||||||
|
var options = new SiteRuntimeOptions
|
||||||
|
{
|
||||||
|
TagSubscribeRetryIntervalMs = 200,
|
||||||
|
TagSubscribeRetryMaxIntervalMs = 800,
|
||||||
|
TagSubscribeInFlightCeilingMs = 60_000, // guard never expires; only replies drive retries
|
||||||
|
};
|
||||||
|
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||||
|
|
||||||
|
var req = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||||
|
var gaps = new List<TimeSpan>();
|
||||||
|
for (var i = 0; i < 3; i++)
|
||||||
|
{
|
||||||
|
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||||||
|
false, "gateway busy", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||||
|
req = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
|
||||||
|
gaps.Add(sw.Elapsed);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Expected nominal gaps: 400, 800(cap), 800(cap) — each ±20 % jitter and scheduler slack.
|
||||||
|
Assert.InRange(gaps[0].TotalMilliseconds, 300, 560);
|
||||||
|
Assert.InRange(gaps[1].TotalMilliseconds, 600, 1100);
|
||||||
|
Assert.InRange(gaps[2].TotalMilliseconds, 600, 1100);
|
||||||
|
|
||||||
|
// Success resets: no further request follows a confirmed subscription.
|
||||||
|
instance.Tell(new SubscribeTagsResponse(req.CorrelationId, req.InstanceUniqueName,
|
||||||
|
true, null, DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
dclProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(1200));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void FailureReply_ReissuesEvenThoughGuardHasNotExpired()
|
||||||
|
{
|
||||||
|
var options = new SiteRuntimeOptions
|
||||||
|
{
|
||||||
|
TagSubscribeRetryIntervalMs = 200,
|
||||||
|
TagSubscribeRetryMaxIntervalMs = 200,
|
||||||
|
TagSubscribeInFlightCeilingMs = 60_000,
|
||||||
|
};
|
||||||
|
var (instance, dclProbe) = CreateInstanceWithDataSourcedAttribute(options);
|
||||||
|
|
||||||
|
var first = dclProbe.ExpectMsg<SubscribeTagsRequest>();
|
||||||
|
instance.Tell(new SubscribeTagsResponse(first.CorrelationId, first.InstanceUniqueName,
|
||||||
|
false, "Unknown connection: conn-1", DateTimeOffset.UtcNow), dclProbe.Ref);
|
||||||
|
|
||||||
|
// A definite failure clears the in-flight slot, so the watchdog re-sends.
|
||||||
|
var retry = dclProbe.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(3));
|
||||||
|
Assert.NotEqual(first.CorrelationId, retry.CorrelationId);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user