fix(mqtt): a reconnect that re-establishes nothing tears the session down
Driver.Mqtt-007 (Low, error handling). OnReconnectedAsync throws when a re-subscribe is granted zero filters — MqttConnection's contract is that a throwing Reconnected subscriber tears the fresh session down and retries under backoff, because a connected-but-deaf driver is the one outcome this path must never serve. But when the live subscribed topics resolved to zero filters in the first place it logged a warning and returned, reaching the same "nothing was re-established" state by a different route and giving it the opposite answer. Throw there too, and say so in the remarks so the type states ONE rule. It costs no other tag — every live topic is in the set that resolved to nothing, so there is nothing left flowing to protect, which is the whole reason the partial- rejection case is exempt — and it is not permanent: the retry runs against whatever Register has published by then. Test reaches the state the only way production can: Register prunes _subscribedTopics of the topics its new table no longer carries, but a SUBSCRIBE already in flight records its own topic AFTER that prune ran, so a deploy landing inside the round-trip leaves a live topic no authored tag resolves to. Paired with the control that a reconnect with no live subscriptions at all stays a no-op — throwing there would tear down the session of every driver whose tags have not been subscribed yet.
This commit is contained in:
@@ -397,6 +397,15 @@ public sealed class MqttSubscriptionManager
|
|||||||
/// rest keep flowing. That is the plan's "SUBACK failure → per-ref Bad, not hang", scoped
|
/// rest keep flowing. That is the plan's "SUBACK failure → per-ref Bad, not hang", scoped
|
||||||
/// to the case where it does not cost the whole driver.
|
/// to the case where it does not cost the whole driver.
|
||||||
/// </para>
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b>Live topics resolving to zero filters is a total failure too</b> — it re-establishes
|
||||||
|
/// nothing, so it throws by the same rule rather than warning and returning. It costs no
|
||||||
|
/// other tag (every live topic is in the set that resolved to nothing) and it is not
|
||||||
|
/// permanent, because the retry runs against whatever <see cref="Register"/> has published
|
||||||
|
/// by then. It is reachable only when a deploy lands inside a subscribe round-trip:
|
||||||
|
/// <see cref="Register"/> prunes <c>_subscribedTopics</c> of the topics its new table no
|
||||||
|
/// longer carries, but an in-flight SUBSCRIBE records its own topic after that prune ran.
|
||||||
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
/// <param name="cancellationToken">The connection's lifetime token; cancelled by its dispose.</param>
|
/// <param name="cancellationToken">The connection's lifetime token; cancelled by its dispose.</param>
|
||||||
/// <returns>A task that completes when the re-subscribe has been answered.</returns>
|
/// <returns>A task that completes when the re-subscribe has been answered.</returns>
|
||||||
@@ -421,24 +430,21 @@ public sealed class MqttSubscriptionManager
|
|||||||
var filters = BuildFilters(live, _defaultQos);
|
var filters = BuildFilters(live, _defaultQos);
|
||||||
if (filters.Count == 0)
|
if (filters.Count == 0)
|
||||||
{
|
{
|
||||||
// Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. If any
|
// Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. Live
|
||||||
// topic was live, resolving zero filters for it is an anomaly and must be loud.
|
// topics resolving to zero filters re-establishes nothing, which is the total-failure case
|
||||||
|
// by a different route, so it gets the total-failure answer — see the remarks.
|
||||||
if (subscribed.Count > 0)
|
if (subscribed.Count > 0)
|
||||||
{
|
{
|
||||||
_logger?.LogWarning(
|
throw new InvalidOperationException(
|
||||||
"MQTT driver '{DriverId}': reconnect resolved NO filters for {TopicCount} live topic(s) "
|
$"MQTT driver '{_driverId}': reconnect resolved NO filters for {subscribed.Count} live "
|
||||||
+ "({Topics}); those subscriptions are not being re-established.",
|
+ $"topic(s) ({string.Join(", ", subscribed)}); tearing the session down rather than "
|
||||||
_driverId,
|
+ "serving a connection that receives nothing.");
|
||||||
subscribed.Count,
|
|
||||||
string.Join(", ", subscribed));
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
_logger?.LogDebug(
|
|
||||||
"MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.",
|
|
||||||
_driverId);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
_logger?.LogDebug(
|
||||||
|
"MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.",
|
||||||
|
_driverId);
|
||||||
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -583,6 +583,51 @@ public sealed class MqttSubscriptionManagerTests
|
|||||||
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good);
|
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// <b>Driver.Mqtt-007.</b> Live subscribed topics that resolve to ZERO filters re-establish
|
||||||
|
/// nothing, which is the same outcome as a total SUBACK failure and must get the same answer:
|
||||||
|
/// throw, so <see cref="MqttConnection"/> tears the fresh session down and retries rather than
|
||||||
|
/// serving a connected-but-deaf driver. A warning was not enough — the whole point of the type's
|
||||||
|
/// split contract is that "no subscription was re-established" is fatal to the session.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The state is reached the only way production can reach it: <see cref="MqttSubscriptionManager.Register"/>
|
||||||
|
/// prunes <c>_subscribedTopics</c> of the topics its new table no longer carries, but a SUBSCRIBE
|
||||||
|
/// already in flight records its topic <i>after</i> that prune has run — so a redeploy landing
|
||||||
|
/// inside the subscribe round-trip leaves a live topic no authored tag resolves to.
|
||||||
|
/// </remarks>
|
||||||
|
[Fact]
|
||||||
|
public async Task OnReconnectedAsync_LiveTopicResolvingNoFilter_Throws_RatherThanServingADeafSession()
|
||||||
|
{
|
||||||
|
var transport = new FakeTransport();
|
||||||
|
var mgr = NewManager(transport);
|
||||||
|
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
|
||||||
|
|
||||||
|
transport.DuringSubscribe = () => mgr.Register([]); // the redeploy lands mid-round-trip
|
||||||
|
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
||||||
|
transport.DuringSubscribe = null;
|
||||||
|
|
||||||
|
await Should.ThrowAsync<InvalidOperationException>(
|
||||||
|
() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The falsifying control: a reconnect with no live subscriptions at all re-establishes nothing
|
||||||
|
/// because there is nothing to establish, and must stay a no-op. Throwing here would tear down
|
||||||
|
/// the session of every driver whose tags have not been subscribed yet.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task OnReconnectedAsync_NoLiveSubscriptions_IsANoOp()
|
||||||
|
{
|
||||||
|
var transport = new FakeTransport();
|
||||||
|
var mgr = NewManager(transport);
|
||||||
|
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
|
||||||
|
|
||||||
|
await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
|
||||||
|
|
||||||
|
transport.Calls.Count.ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------------
|
||||||
// C1 — wildcard-authored tags on the paths that reconstruct state from a FILTER string.
|
// C1 — wildcard-authored tags on the paths that reconstruct state from a FILTER string.
|
||||||
// Delivery matches an incoming topic (concrete); reconnect + degrade key off the subscribed
|
// Delivery matches an incoming topic (concrete); reconnect + degrade key off the subscribed
|
||||||
@@ -898,10 +943,18 @@ public sealed class MqttSubscriptionManagerTests
|
|||||||
/// <summary>When set, the whole subscribe call throws instead of answering.</summary>
|
/// <summary>When set, the whole subscribe call throws instead of answering.</summary>
|
||||||
public Exception? Throw { get; set; }
|
public Exception? Throw { get; set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Runs inside the subscribe round-trip, so a test can land a concurrent
|
||||||
|
/// <see cref="MqttSubscriptionManager.Register"/> in the window between its prune and this
|
||||||
|
/// call recording its topic as live.
|
||||||
|
/// </summary>
|
||||||
|
public Action? DuringSubscribe { get; set; }
|
||||||
|
|
||||||
public Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
|
public Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
|
||||||
IReadOnlyList<MqttTopicSubscription> filters, CancellationToken cancellationToken)
|
IReadOnlyList<MqttTopicSubscription> filters, CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
Calls.Add(filters);
|
Calls.Add(filters);
|
||||||
|
DuringSubscribe?.Invoke();
|
||||||
if (Throw is not null) throw Throw;
|
if (Throw is not null) throw Throw;
|
||||||
IReadOnlyList<MqttSubscribeOutcome> outcomes =
|
IReadOnlyList<MqttSubscribeOutcome> outcomes =
|
||||||
[.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, !Reject(f.Topic), Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1"))];
|
[.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, !Reject(f.Topic), Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1"))];
|
||||||
|
|||||||
Reference in New Issue
Block a user