fix(#3321): step 3 — a disposed stream releases its hub

src/MeshWeaver.Data/GenericUnpartitionedDataSource.cs

@@ -309,13 +309,27 @@ This with
 
     /// <summary>
     /// A task that completes once every created stream's hub has started.
+    ///
+    /// <para>🚨 A stream that has RELEASED its hub contributes nothing to wait for (#3321 step 3):
+    /// the hub is gone, so its <c>Started</c> task cannot be reached, let alone settle.</para>
+    ///
+    /// <para>🚨 <b>And the guard is PRESENCE (<c>HubIfHeld</c>), never liveness
+    /// (<c>TryGetHub</c>) — measured.</b> Writing this with <c>TryGetHub()</c> broke
+    /// <c>DataContextFaultedInitBeforeStreamHubBoundTest</c>: a stream whose initial load FAULTS is
+    /// not <c>IsUsable</c>, so it was excluded from the WhenAll, this task completed
+    /// SUCCESSFULLY, and a hub whose data source had thrown went on answering requests as though
+    /// nothing had happened. A faulted stream's <c>Started</c> task is the very thing that must
+    /// still be awaited — it is how the fault reaches <c>DataContext</c>'s gate.</para>
     /// </summary>
     public Task Initialized
     {
         get
         {
             lock (Streams)
-                return Task.WhenAll(Streams.Values.Select(s => s.Hub.Started));
+                return Task.WhenAll(Streams.Values
+                    .Select(s => s.HubIfHeld())
+                    .Where(h => h is not null)
+                    .Select(h => h!.Started));
         }
     }
     /// <summary>

src/MeshWeaver.Data/ISynchronizationStream.cs

@@ -86,14 +86,22 @@ ISynchronizationStream Reduce(
 
     /// <summary>The message hub associated with this stream.
     ///
-    /// <para>🚨 <b>Non-nullable in the contract, but not in life.</b> A stream whose owner has been
-    /// torn down keeps answering this property, and the hub it hands back may already be winding
-    /// down — dereferencing it is how a recycle window produced an NRE inside
-    /// <c>LayoutAreaHost</c>'s constructor that escaped to the subscriber as a TERMINAL
-    /// <c>DeliveryFailure</c> (Systemorph/MeshWeaver#3321). On any path that can run concurrently
-    /// with teardown, prefer <see cref="SynchronizationStreamLiveness.TryGetHub"/>, which answers
-    /// <c>null</c> instead, or gate on
-    /// <see cref="SynchronizationStreamLiveness.IsUsable(ISynchronizationStream?)"/>.</para>
+    /// <para>🚨 <b>Non-nullable in the contract, but not in life — and since #3321 step 3 it is
+    /// ABSENT on a stream that has let its hub go.</b> A disposed stream, and a stream whose hosted
+    /// sub-hub was killed by its parent's teardown, both RELEASE this reference: that release is
+    /// what reclaims the <c>stream → dead hub → resolved state</c> graph that held ~580 MB of a
+    /// production replica's heap. The declaration stays non-nullable because these assemblies ship
+    /// as packages and changing it would break every downstream implementer.</para>
+    ///
+    /// <para><b>So never dereference this on a path that can run concurrently with teardown.</b> Use
+    /// <see cref="SynchronizationStreamLiveness.TryGetHub"/>, which answers <c>null</c>, or gate on
+    /// <see cref="SynchronizationStreamLiveness.IsUsable(ISynchronizationStream?)"/>. Where the
+    /// surface has no absent value to return — a patch reducer, whose signature is
+    /// <c>… → ChangeItem&lt;T&gt;</c> — use
+    /// <see cref="SynchronizationStreamLiveness.RequireHub"/>, which refuses with the TRANSIENT
+    /// <c>HubDisposingException</c> rather than an NRE. A raw dereference here is how a recycle
+    /// window produced an NRE inside <c>LayoutAreaHost</c>'s constructor that escaped to the
+    /// subscriber as a TERMINAL <c>DeliveryFailure</c>.</para>
     /// </summary>
     IMessageHub Hub { get; }
     /// <summary>The hub that hosts the underlying data source backing this stream.</summary>

src/MeshWeaver.Data/Persistence/PartitionedHubDataSource.cs

@@ -98,13 +98,27 @@ protected override ISynchronizationStream<EntityStore> CreateStream(StreamIdenti
                 .Synchronize()
                 .Subscribe(changes =>
                 {
+                    // 🚨 #3321 step 3 — same shape as WorkspaceStreams' combined subscription: a
+                    // callback in flight when the stream is torn down finds `ret.Hub` released.
+                    // Dropping the frame is what `ret.OnNext` would do anyway on a released stream,
+                    // so it is the answer this void callback already models. PRESENCE, not liveness
+                    // (HubIfHeld, not TryGetHub): this guard must match OnNext's own
+                    // `isDisposed || Hub is null` exactly, so it refuses precisely when the
+                    // emission would be dropped and never a frame that would still have landed.
+                    if (ret.HubIfHeld() is not { } retHub)
+                    {
+                        Logger.LogDebug(
+                            "PartitionedHubDataSource: combined stream {StreamId} is torn down — "
+                            + "dropping this frame", ret.StreamId);
+                        return;
+                    }
                     if (!isInitialized)
                     {
                         var initialStore = changes
                             .Where(c => c.Value != null)
                             .Aggregate(new EntityStore(), (acc, change) => acc.Merge(change.Value!));
 
-                        ret.OnNext(new ChangeItem<EntityStore>(initialStore, ret.StreamId, ret.Hub.Version));
+                        ret.OnNext(new ChangeItem<EntityStore>(initialStore, ret.StreamId, retHub.Version));
 
                         foreach (var change in changes)
                             processedChangeItems.Add(change);
@@ -120,7 +134,7 @@ protected override ISynchronizationStream<EntityStore> CreateStream(StreamIdenti
                                 .Where(c => c.Value != null)
                                 .Aggregate(new EntityStore(), (acc, change) => acc.Merge(change.Value!));
 
-                            ret.OnNext(new ChangeItem<EntityStore>(updatedStore, ret.StreamId, ret.Hub.Version)
+                            ret.OnNext(new ChangeItem<EntityStore>(updatedStore, ret.StreamId, retHub.Version)
                             {
                                 ChangedBy = string.Join(", ", newChanges.Select(c => c.ChangedBy).Where(cb => !string.IsNullOrEmpty(cb))),
                                 Updates = newChanges.SelectMany(c => c.Updates).ToArray(),

src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs

@@ -576,16 +576,37 @@ internal static ISynchronizationStream CreateExternalClient<TReduced, TReference
         // Belt-and-suspenders: dispose the subscription when the HUB tears down too (idempotent).
         hub.RegisterForDisposal(observeSubscription);
 
-        reduced.RegisterForDisposal(
-            reduced.Hub.Register<UnsubscribeRequest>(
-                delivery =>
-                {
-                    reduced.DeliverMessage(delivery);
-                    return delivery.Forwarded();
-                },
-                d => reduced.StreamId.Equals(d.Message.StreamId)
-            )
-        );
+        // 🚨 The stream's hub is resolved ONCE here and reused for both owner-protocol
+        // registrations below (#3321 step 3). `reduced` was created a few lines up, but its host
+        // can be torn down underneath this setup — and since step 3 a RELEASED stream answers
+        // `Hub` with nothing, so `reduced.Hub.Register(…)` would NRE while EVALUATING the
+        // argument, i.e. before RegisterForDisposal could apply its own already-dead guard. Not
+        // registering is the honest "no": a handler on a hub that is gone can never fire, and
+        // RegisterForDisposal would have disposed the registration on the spot anyway. Nothing
+        // else in this method is skipped — the keep-alive and resubscribe arms below are
+        // independently guarded and must still be wired for a stream that recovers.
+        //
+        // PRESENCE (HubIfHeld), not liveness (TryGetHub): RegisterForDisposal's own guard is
+        // `isDisposed || Hub is null`, and this must match it — refusing to wire the owner
+        // protocol on a stream that is merely winding down would drop the UnsubscribeRequest and
+        // StreamEndedEvent handling for a stream that is still going to serve.
+        var reducedHub = reduced.HubIfHeld();
+        if (reducedHub is null)
+            logger.LogDebug(
+                "Stream {StreamId}: released its hub before the owner-protocol handlers could be "
+                + "registered — nothing to wire (#3321)", reduced.StreamId);
+
+        else
+            reduced.RegisterForDisposal(
+                reducedHub.Register<UnsubscribeRequest>(
+                    delivery =>
+                    {
+                        reduced.DeliverMessage(delivery);
+                        return delivery.Forwarded();
+                    },
+                    d => reduced.StreamId.Equals(d.Message.StreamId)
+                )
+            );
 
         // 🚨 THE OWNER ANNOUNCING THAT OUR SERVER-SIDE HALF HAS ENDED (#2191).
         //
@@ -605,27 +626,28 @@ internal static ISynchronizationStream CreateExternalClient<TReduced, TReference
         //
         // NOT a watchdog — no timer, no poll, no backoff. One re-ask per real end event, and a
         // re-ask can never CAUSE an end, so there is nothing here to amplify.
-        reduced.RegisterForDisposal(
-            reduced.Hub.Register<StreamEndedEvent>(
-                delivery =>
-                {
-                    // Debug, not Information: Resubscribe already logs the re-ask at Information
-                    // WITH this reason string, so an Information line here would double the Loki
-                    // volume of one event.
-                    logger.LogDebug(
-                        "Stream {StreamId}: owner {Owner} ended the server-side subscription — re-asking once its teardown settles.",
-                        reduced.StreamId, owner);
-                    // No activation identity on this carrier — the owner ANNOUNCED the end rather
-                    // than NACKing, so there is no NACK text to read one from. A null tag is
-                    // treated as "a distinct activation" by the budget below, i.e. exactly the
-                    // pre-#2986 accounting: never guessed, never free.
-                    rejectedByRecycle.OnNext(new RecycleRejection(
-                        "ended our server-side subscription", null));
-                    return delivery.Processed();
-                },
-                d => reduced.StreamId.Equals(d.Message.StreamId)
-            )
-        );
+        if (reducedHub is not null)
+            reduced.RegisterForDisposal(
+                reducedHub.Register<StreamEndedEvent>(
+                    delivery =>
+                    {
+                        // Debug, not Information: Resubscribe already logs the re-ask at
+                        // Information WITH this reason string, so an Information line here would
+                        // double the Loki volume of one event.
+                        logger.LogDebug(
+                            "Stream {StreamId}: owner {Owner} ended the server-side subscription — re-asking once its teardown settles.",
+                            reduced.StreamId, owner);
+                        // No activation identity on this carrier — the owner ANNOUNCED the end
+                        // rather than NACKing, so there is no NACK text to read one from. A null
+                        // tag is treated as "a distinct activation" by the budget below, i.e.
+                        // exactly the pre-#2986 accounting: never guessed, never free.
+                        rejectedByRecycle.OnNext(new RecycleRejection(
+                            "ended our server-side subscription", null));
+                        return delivery.Processed();
+                    },
+                    d => reduced.StreamId.Equals(d.Message.StreamId)
+                )
+            );
 
         reduced.RegisterForDisposal(
             new AnonymousDisposable(
@@ -1802,17 +1824,60 @@ fromWorkspace as ISynchronizationStream<TReduced>
     /// <param name="currentJson">The current state serialized as JSON.</param>
     /// <param name="patch">The JSON patch to apply, or null for a full update.</param>
     /// <param name="changedBy">Identifier of the principal that made the change.</param>
-    /// <returns>The resulting change item, or null when no patch function is registered.</returns>
+    /// <returns>The resulting change item, or null when no patch function is registered — or when
+    /// the stream has released its hub, which is the same "nothing to derive" answer.</returns>
     public static ChangeItem<TReduced>? ToChangeItem<TReduced>(
         this ISynchronizationStream<TReduced> stream,
         TReduced currentState,
         JsonElement currentJson,
         JsonPatch? patch,
         string changedBy)
     {
-        return stream.ReduceManager.PatchFunction?.Invoke(stream, currentState, currentJson, patch, changedBy);
+        try
+        {
+            return stream.ReduceManager.PatchFunction?.Invoke(stream, currentState, currentJson, patch, changedBy);
+        }
+        catch (HubDisposingException)
+        {
+            // 🚨 THE ONE GATEWAY into every patch reducer, so this is the one place that has to
+            // translate their refusal (#3321 step 3). A reducer's signature is `… → ChangeItem<T>`:
+            // it must return a change, so when the stream has RELEASED its hub — step 3 drops the
+            // reference on disposal and on the hub's own death, which is what reclaims the leaked
+            // `stream → dead hub → resolved state` graph — its only channel is
+            // `RequireHub()`'s transient throw.
+            //
+            // This is NOT a swallow. `null` is this method's OWN modelled answer for "no change was
+            // derived" (it is what a stream with no PatchFunction registered already returns), and
+            // both callers handle it: SynchronizationStream.BuildChangeItem falls through to its
+            // type-registry path, and UpdateStream falls back to deserializing the patched JSON
+            // through Host.JsonSerializerOptions. Only THIS exception type is caught, by design —
+            // anything else a reducer throws is a real fault and still propagates.
+            //
+            // Debug, not Warning: the window is a teardown race on a stream that is already gone,
+            // it produces no user-visible effect (a dead stream's store is completed, so the change
+            // had nowhere to land), and it can fan out across every reduced stream of one dying hub.
+            LogPatchSkippedOnReleasedStream(stream);
+            return null;
+        }
     }
 
+    /// <summary>
+    /// Best-effort Debug line for the skip above.
+    ///
+    /// <para>🚨 Resolved through <see cref="GetLogger"/> rather than a bare
+    /// <c>ServiceProvider.GetService&lt;ILoggerFactory&gt;()</c>, and that matters here more than
+    /// anywhere else in this file: a stream whose hub was just released is very often one whose HOST
+    /// is winding down too, and Autofac throws <c>ObjectDisposedException</c> from a closed scope. A
+    /// raw resolution would turn a benign "no patch derived" into a fault raised BY THE DIAGNOSTIC,
+    /// on the exact teardown path this code exists to make quiet. <c>GetLogger</c> already answers
+    /// that with a <c>NullLogger</c> — losing the line is the correct price, and
+    /// <c>SynchronizationStream.ResolveLogger</c> carries the same guard for the same reason.</para>
+    /// </summary>
+    private static void LogPatchSkippedOnReleasedStream(ISynchronizationStream stream)
+        => GetLogger(stream.Host.ServiceProvider).LogDebug(
+            "[SYNC_STREAM] Patch reduction skipped for {StreamId} ({Reference}) — the stream "
+            + "released its hub before the reducer ran (#3321)", stream.StreamId, stream.Reference);
+
 
     /// <summary>
     /// Converts a JSON patch over a whole entity-store projection into per-entity