fix(#2756): a transient unserved verdict must not evict the owner's server-side stream

src/MeshWeaver.Data/DataExtensions.cs

@@ -1838,17 +1838,25 @@ private static IMessageDelivery HandleGetDomainTypesRequest(IMessageHub hub, IMe
     /// delivery passes through unchanged for whoever else handles it (the request/response callback
     /// already ran; it is first in the rule chain).</para>
     ///
-    /// <para>🚨 <b>The <c>ErrorType == NotFound</c> half of this gate was removed deliberately.</b>
-    /// It was redundant belt-and-braces from the era when the router's ONLY unserved verdict came
-    /// from the stream leg's subscriber probe. Since
-    /// <c>RoutingGrain.AnswerPodHubNotHere</c> the router also stamps the verdict on the
-    /// TRANSIENT <see cref="ErrorType.ShuttingDown"/> — "no silo serves this hub right now" reached
-    /// in one hop through the grain directory, the answer that replaced the stream publish
-    /// (<c>Doc/Architecture/DurableStreamsViaMeshNodes</c>). Keeping the NotFound test would have
-    /// made that answer inert here and silently re-opened this very leak for every dead circuit in
-    /// the fleet. The two verdicts are complementary, not contradictory: the SUBSCRIBER rides
-    /// <c>ShuttingDown</c> out and re-asks, while the OWNER drops the server-side half it can no
-    /// longer push to — which is exactly the recovery described below.</para>
+    /// <para>🚨 <b>A TRANSIENT verdict must NOT evict — issue #2756.</b> The stamp says "no silo
+    /// serves this address"; the <see cref="ErrorType"/> beside it says whether that is a fact
+    /// about a DEAD process or about a moment in time. Since <c>RoutingGrain.AnswerPodHubNotHere</c>
+    /// the router also stamps the verdict on <see cref="ErrorType.ShuttingDown"/> — reached in one
+    /// hop through Orleans' grain directory when the owner is mid-roll or its pod-hub claim has not
+    /// landed yet (<c>Doc/Architecture/DurableStreamsViaMeshNodes</c>). That one is explicitly
+    /// recoverable, and <c>JsonSynchronizationStream</c> is built to ride it out and RE-ARM rather
+    /// than re-subscribe. So evicting on it destroys the server-side half of a subscription whose
+    /// other half is deliberately sitting still — the owner throws the stream away while the
+    /// subscriber waits for it. #2745 briefly gated on the stamp alone and turned main red on
+    /// <c>ObservableQueryTests.ObserveQuery_EmitsRemovedOnDeletedNode</c>, where both halves live
+    /// in one process and the eviction wins the race against the mirror.</para>
+    ///
+    /// <para>The leak this handler exists to close (#2426/#2546) is a subscriber whose PROCESS is
+    /// GONE — the terminal verdict. A transient one is not that, and answering it with an
+    /// irreversible eviction trades a bounded wait for a lost subscription. Hence: the stamp
+    /// decides whether the delivery is OURS to act on, and the verdict decides whether the right
+    /// action is to evict or to leave the stream alone. Both halves are load-bearing, and each is
+    /// pinned by its own test in <c>UnservedVerdictEvictionTest</c>.</para>
     ///
     /// <para>Evict-only, and terminal per NACK: nothing here retries, re-probes or resubscribes. A
     /// subscriber that was in fact alive but unreachable loses only its server-side stream, and
@@ -1861,6 +1869,14 @@ private static IMessageDelivery HandleTargetUnservedFailure(
         var failure = delivery.Message;
         if (!failure.TargetUnserved)
             return delivery;
+        // 🚨 TRANSIENT ⇒ leave the stream alone (#2756). ShuttingDown is the platform's
+        // "come back and re-ask" verdict — JsonSynchronizationStream rides it out and re-arms
+        // instead of re-subscribing — so evicting here would dispose the server-side half of a
+        // subscription that is deliberately waiting for it. A NEW transient ErrorType must be added
+        // to this test, not left to fall through: the default here is to evict, and eviction is
+        // irreversible for a subscriber that never re-asks.
+        if (failure.ErrorType == ErrorType.ShuttingDown)
+            return delivery;
         var deadSubscriber = failure.Delivery?.Target;
         if (deadSubscriber is null || hub.GetWorkspace() is not Workspace workspace)
             return delivery;

test/MeshWeaver.Data.Test/UnservedVerdictEvictionTest.cs

@@ -23,21 +23,26 @@ namespace MeshWeaver.Data.Test;
 /// <see cref="DeliveryFailure.TargetUnserved"/> verdict reaching the owner, which disposes that
 /// subscriber's server-side stream (20,718 error lines in 3 h on memex-cloud before it did).</para>
 ///
-/// <para>🚨 <b>Why this test exists NOW.</b> That verdict used to have exactly one producer — the
-/// stream leg's no-live-subscriber refusal, which stamps it beside
-/// <see cref="ErrorType.NotFound"/> — so the handler could afford a redundant
-/// <c>ErrorType == NotFound</c> test beside the stamp. Since
-/// <c>RoutingGrain.AnswerPodHubNotHere</c> the router reaches the same conclusion ONE HOP EARLIER,
-/// through Orleans' grain directory, and reports it as the TRANSIENT
-/// <see cref="ErrorType.ShuttingDown"/> — because a hub whose owner is mid-roll must not have its
-/// subscriber's mirrors torn down. Had the ErrorType test survived, that verdict would have been
-/// inert here and every dead circuit in the fleet would have leaked its server-side streams again.
-/// The two facts are complementary, not contradictory: the SUBSCRIBER rides the transient verdict
-/// out and re-asks; the OWNER drops the half it can no longer push to.</para>
+/// <para>🚨 <b>The stamp says WHOSE delivery this is; the ErrorType says WHAT TO DO — #2756.</b>
+/// The verdict used to have exactly one producer — the stream leg's no-live-subscriber refusal,
+/// which stamps it beside <see cref="ErrorType.NotFound"/>. Since
+/// <c>RoutingGrain.AnswerPodHubNotHere</c> the router reaches a similar conclusion ONE HOP EARLIER,
+/// through Orleans' grain directory, but reports it as the TRANSIENT
+/// <see cref="ErrorType.ShuttingDown"/> — the owner may simply be mid-roll, or waiting for its
+/// pod-hub claim to land.</para>
+///
+/// <para>#2745 gated the eviction on the STAMP ALONE, reasoning the two verdicts were
+/// complementary: the subscriber rides the transient one out while the owner drops its half. They
+/// are not. <c>JsonSynchronizationStream</c> rides <c>ShuttingDown</c> out by RE-ARMING rather than
+/// re-subscribing, so an owner that evicts on it disposes the server-side half of a subscription
+/// whose other half is deliberately sitting still. That turned main RED on
+/// <c>ObservableQueryTests.ObserveQuery_EmitsRemovedOnDeletedNode</c>, where both halves share one
+/// process and the eviction wins the race. The leak this handler closes (#2426/#2546) is a
+/// subscriber whose PROCESS IS GONE — the terminal verdict — and only that one may evict.</para>
 ///
 /// <para><b>Fails on unfixed code:</b>
-/// <see cref="ATransientUnservedVerdict_EvictsTheOwnersServerSideStream"/> — with the
-/// <c>ErrorType == NotFound</c> test in place the eviction counter never leaves zero.</para>
+/// <see cref="ATransientUnservedVerdict_LeavesTheOwnersServerSideStreamAlone"/> — with the
+/// stamp-alone gate the transient verdict evicts and the subscription is destroyed.</para>
 ///
 /// <para>See <c>Doc/Architecture/DurableStreamsViaMeshNodes</c> and
 /// <c>Doc/Architecture/ErrorPropagationAndWedges</c>.</para>
@@ -93,6 +98,10 @@ private static DeliveryFailure RouterVerdict(IMessageHub host, Address subscribe
             TargetUnserved = true,
         };
 
+    /// <summary>Marker that the owner writes after the transient verdict — its arrival on the
+    /// client is the proof that the server-side stream was not evicted.</summary>
+    private const string SurvivedTheVerdict = "survived-the-transient-verdict";
+
     private static Task<int> EvictedAsync(Workspace workspace) =>
         Observable.Interval(50.Milliseconds())
             .StartWith(0L)
@@ -103,21 +112,45 @@ private static Task<int> EvictedAsync(Workspace workspace) =>
             .ToTask(TestContext.Current.CancellationToken);
 
     /// <summary>
-    /// 🚨 THE PIN. The transient verdict the retired stream leg's replacement produces must still
-    /// end the fan-out at its source.
+    /// 🚨 THE PIN (#2756). A TRANSIENT unserved verdict must leave the server-side stream intact:
+    /// the subscriber rides <c>ShuttingDown</c> out by re-arming, not by re-subscribing, so an
+    /// eviction here destroys the half it is waiting for.
     /// </summary>
     [HubFact]
-    public async Task ATransientUnservedVerdict_EvictsTheOwnersServerSideStream()
+    public async Task ATransientUnservedVerdict_LeavesTheOwnersServerSideStreamAlone()
     {
         var (host, client, hostWorkspace) = await ServingClientAsync();
 
         host.Post(RouterVerdict(host, client.Address, ErrorType.ShuttingDown),
             o => o.WithTarget(host.Address));
 
-        (await EvictedAsync(hostWorkspace)).Should().BeGreaterThanOrEqualTo(1,
-            "the eviction gate is the router's TargetUnserved STAMP — the only component that asks "
-            + "the cluster — and never the ErrorType beside it. Gating on NotFound made the pod-hub "
-            + "refusal inert and left every dead circuit fanning out forever (#2426/#2546)");
+        // 🚨 The assertion is a POSITIVE signal, not an elapsed window: change the data on the
+        // OWNER and require the change to reach the client's mirror. That can only happen over the
+        // server-side stream this verdict must not have evicted — so the wait is on the property
+        // itself, and a regression fails as a timeout rather than passing before the verdict was
+        // even handled. (A counter check alone would do the latter: it can be read as zero long
+        // before the delivery reaches the handler.)
+        var hostUnits = await hostWorkspace.GetStream<BusinessUnit>()!
+            .Should().Within(10.Seconds()).Emit();
+        var updated = hostUnits!.First() with { DisplayName = SurvivedTheVerdict };
+        host.Post(new DataChangeRequest { Updates = [updated] });
+
+        var clientWorkspace = client.ServiceProvider.GetRequiredService<IWorkspace>();
+        var mirrored = await clientWorkspace
+            .GetObservable<BusinessUnit>(updated.SystemName)
+            .Should().Within(10.Seconds())
+            .Match(x => x!.DisplayName == SurvivedTheVerdict);
+        mirrored.Should().Be(updated,
+            "ShuttingDown is the platform's transient verdict — JsonSynchronizationStream rides it "
+            + "out and RE-ARMS rather than re-subscribing. Evicting on it disposes the server-side "
+            + "half of a live subscription whose other half is deliberately waiting, so the owner's "
+            + "next change never arrives (#2756, which turned main red on "
+            + "ObservableQueryTests.ObserveQuery_EmitsRemovedOnDeletedNode)");
+
+        // Only now is the counter meaningful: the update above proves the verdict was delivered
+        // and handled, so a zero here is a decision, not a race.
+        hostWorkspace.ClientSubscriptionsEvicted.Should().Be(0,
+            "no eviction may be recorded for a transient verdict");
     }
 
     /// <summary>