diff --git a/src/MeshWeaver.Data/GenericUnpartitionedDataSource.cs b/src/MeshWeaver.Data/GenericUnpartitionedDataSource.cs index b93b4bd8fe..567a20aa28 100644 --- a/src/MeshWeaver.Data/GenericUnpartitionedDataSource.cs +++ b/src/MeshWeaver.Data/GenericUnpartitionedDataSource.cs @@ -309,13 +309,27 @@ This with /// /// A task that completes once every created stream's hub has started. + /// + /// 🚨 A stream that has RELEASED its hub contributes nothing to wait for (#3321 step 3): + /// the hub is gone, so its Started task cannot be reached, let alone settle. + /// + /// 🚨 And the guard is PRESENCE (HubIfHeld), never liveness + /// (TryGetHub) β€” measured. Writing this with TryGetHub() broke + /// DataContextFaultedInitBeforeStreamHubBoundTest: a stream whose initial load FAULTS is + /// not IsUsable, 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 Started task is the very thing that must + /// still be awaited β€” it is how the fault reaches DataContext's gate. /// 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)); } } /// diff --git a/src/MeshWeaver.Data/ISynchronizationStream.cs b/src/MeshWeaver.Data/ISynchronizationStream.cs index eb4f224e0b..62d951fcbf 100644 --- a/src/MeshWeaver.Data/ISynchronizationStream.cs +++ b/src/MeshWeaver.Data/ISynchronizationStream.cs @@ -86,14 +86,22 @@ ISynchronizationStream Reduce( /// The message hub associated with this stream. /// - /// 🚨 Non-nullable in the contract, but not in life. 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 - /// LayoutAreaHost's constructor that escaped to the subscriber as a TERMINAL - /// DeliveryFailure (Systemorph/MeshWeaver#3321). On any path that can run concurrently - /// with teardown, prefer , which answers - /// null instead, or gate on - /// . + /// 🚨 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. 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 stream β†’ dead hub β†’ resolved state 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. + /// + /// So never dereference this on a path that can run concurrently with teardown. Use + /// , which answers null, or gate on + /// . Where the + /// surface has no absent value to return β€” a patch reducer, whose signature is + /// … β†’ ChangeItem<T> β€” use + /// , which refuses with the TRANSIENT + /// HubDisposingException rather than an NRE. A raw dereference here is how a recycle + /// window produced an NRE inside LayoutAreaHost's constructor that escaped to the + /// subscriber as a TERMINAL DeliveryFailure. /// IMessageHub Hub { get; } /// The hub that hosts the underlying data source backing this stream. diff --git a/src/MeshWeaver.Data/Persistence/PartitionedHubDataSource.cs b/src/MeshWeaver.Data/Persistence/PartitionedHubDataSource.cs index bfd3f363ae..e4c92b8c1b 100644 --- a/src/MeshWeaver.Data/Persistence/PartitionedHubDataSource.cs +++ b/src/MeshWeaver.Data/Persistence/PartitionedHubDataSource.cs @@ -98,13 +98,27 @@ protected override ISynchronizationStream 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(initialStore, ret.StreamId, ret.Hub.Version)); + ret.OnNext(new ChangeItem(initialStore, ret.StreamId, retHub.Version)); foreach (var change in changes) processedChangeItems.Add(change); @@ -120,7 +134,7 @@ protected override ISynchronizationStream CreateStream(StreamIdenti .Where(c => c.Value != null) .Aggregate(new EntityStore(), (acc, change) => acc.Merge(change.Value!)); - ret.OnNext(new ChangeItem(updatedStore, ret.StreamId, ret.Hub.Version) + ret.OnNext(new ChangeItem(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(), diff --git a/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs b/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs index 6b97fcb916..9f791d0480 100644 --- a/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs +++ b/src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs @@ -576,16 +576,37 @@ internal static ISynchronizationStream CreateExternalClient( - 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( + 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( - 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( + 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,7 +1824,8 @@ fromWorkspace as ISynchronizationStream /// The current state serialized as JSON. /// The JSON patch to apply, or null for a full update. /// Identifier of the principal that made the change. - /// The resulting change item, or null when no patch function is registered. + /// 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. public static ChangeItem? ToChangeItem( this ISynchronizationStream stream, TReduced currentState, @@ -1810,9 +1833,51 @@ fromWorkspace as ISynchronizationStream 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`: + // 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; + } } + /// + /// Best-effort Debug line for the skip above. + /// + /// 🚨 Resolved through rather than a bare + /// ServiceProvider.GetService<ILoggerFactory>(), 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 ObjectDisposedException 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. GetLogger already answers + /// that with a NullLogger β€” losing the line is the correct price, and + /// SynchronizationStream.ResolveLogger carries the same guard for the same reason. + /// + 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); + /// /// Converts a JSON patch over a whole entity-store projection into per-entity diff --git a/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs b/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs index b9c6dbe231..0c29015c9f 100644 --- a/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs +++ b/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs @@ -374,6 +374,30 @@ public ChangeItem? Current /// PRE-EXISTING sub-hub, in which case the configuration lambda β€” and therefore /// β€” never runs (and no BuildupAction runs either, so nothing can /// observe the gap). + /// + /// 🚨 RELEASED when the hub dies β€” this field is what leaked + /// (Systemorph/MeshWeaver#3321, step 3 of 3). A heap dump of a 26 h production replica found + /// 1 496 MessageHubs at RunLevel=Dead still reachable, and the ONLY thing holding + /// them was this property: 1 485 SynchronizationStream<MeshNode> + 11 + /// <JsonElement>, one stream per corpse, exactly. Closing a DI scope does not null + /// the fields of an object that outlives it, so each dead hub kept its own Autofac + /// ILifetimeScope and TypeRegistry alive β€” ~390 KB apiece, ~580 MB of the 3.5 GB + /// live heap. is now called from BOTH ends of that lifetime: + /// , and a hook on the sub-hub's own disposal for the far more common case + /// where the PARENT tears the sub-hub down while nobody ever disposes the stream (1 485 of the + /// 1 496). + /// + /// 🚨 So the declaration is non-nullable and the field is not, deliberately. + /// Making it IMessageHub? is a breaking change to a public interface these assemblies + /// ship as packages; what makes it SAFE instead is that every dereference reachable + /// concurrently with teardown goes through + /// (answers null) or + /// (throws the TRANSIENT + /// ), and every read inside this file resolves the field + /// ONCE into a local rather than re-reading it after its own guard. The predecessor of this + /// code nulled the field with neither of those in place and it cost a production NRE inside + /// LayoutAreaHost's constructor that reached the subscriber as a TERMINAL + /// DeliveryFailure. See Doc/Architecture/StreamLivenessAndTheHubReference. /// public IMessageHub Hub { get; private set; } = null!; @@ -383,6 +407,36 @@ public ChangeItem? Current /// private void BindHub(IMessageHub syncHub) => Hub = syncHub; + /// + /// Drops the reference to the synchronization hub β€” the release half of #3321. + /// + /// Idempotent and callable from any thread. It does NOT dispose anything: by the time it + /// runs the hub is either being disposed by us () or has just finished + /// disposing itself (the constructor's hook), so there is nothing left to tear down β€” only a + /// reference to let go of. Everything that could still read the field is already guarded: + /// , the isDisposed || Hub is null arms on + /// / / , and + /// , which has always treated an absent hub as dead β€” so a + /// released stream reports itself unusable through the same predicate every cache already + /// consults. + /// + private void ReleaseHub() + { + var released = Interlocked.Exchange(ref hubReleased, 1) == 0; + if (!released) + return; + // Read the address BEFORE dropping the reference β€” the log line is the only trace that a + // corpse was let go, and it is the counter a heap re-measurement is compared against. + var address = Hub?.Address; + Hub = null!; + logger.LogDebug( + "[SYNC_STREAM] Released hub {Address} from stream {StreamId} β€” the stream no longer " + + "retains it (#3321)", address, StreamId); + } + + /// 0 until has run; CAS'd to 1 exactly once. + private int hubReleased; + /// /// The host of the synchronization stream. /// @@ -648,7 +702,13 @@ public void Update(Func valueUpdate, Action excep var updated = valueUpdate(current); if (updated is null) return null; if (current is not null && Equals(current, updated)) return null; - return BuildChangeItem(current, updated); + // Resolve the hub ONCE, here on the turn (#3321 step 3). The transform runs inside the + // sync hub's UpdateStreamRequest handler, which a Dispose() on another thread can + // overtake β€” and `null` is this delegate's own documented no-op, so a stream that let + // its hub go answers with the absence it already models instead of NRE'ing on the way + // into BuildChangeItem. + if (!TryGetActiveHub(out var hub)) return null; + return BuildChangeItem(hub, current, updated); }, exceptionCallback); /// @@ -670,7 +730,9 @@ public void SetFull(Func valueUpdate, Action exce { var updated = valueUpdate(current); if (updated is null) return null; - return BuildFullChangeItem(current, updated); + // Hub resolved once on the turn β€” see Update above (#3321 step 3). + if (!TryGetActiveHub(out var hub)) return null; + return BuildFullChangeItem(hub, current, updated); }, exceptionCallback); /// @@ -698,10 +760,13 @@ public void SetFull(Func valueUpdate) /// never emitted its content (the DataChangeStreamUpdateTest count-view non-emission). /// /// - private long OwnerVersion() - => Owner.Equals(Host.Address) ? Hub.Version : (Current?.Version ?? 0L); + /// The stream's hub, ALREADY RESOLVED by the caller's guard β€” never + /// re-read from the field here, because a Dispose() on another thread can null it between the + /// guard and this line (#3321 step 3). + private long OwnerVersion(IMessageHub hub) + => Owner.Equals(Host.Address) ? hub.Version : (Current?.Version ?? 0L); - private ChangeItem BuildChangeItem(TStream? current, TStream updated) + private ChangeItem BuildChangeItem(IMessageHub hub, TStream? current, TStream updated) { // 🚨 ChangedBy is the stream-echo-suppression key β€” the identity of the STREAM that // originated the change β€” and it must MATCH the value the echo-suppression filters @@ -719,11 +784,11 @@ private ChangeItem BuildChangeItem(TStream? current, TStream updated) // an empty ChangedBy breaks both filters. ClientId is a non-empty Guid by construction. var changedBy = ClientId; // 🚨 ONLY the owning hub sets Version. Subscriber carries the base it read. - var version = OwnerVersion(); + var version = OwnerVersion(hub); if (current is not null) { - var updatedJson = JsonSerializer.SerializeToElement(updated, Hub.JsonSerializerOptions); + var updatedJson = JsonSerializer.SerializeToElement(updated, hub.JsonSerializerOptions); // 1. PatchFunction (e.g. PatchMeshNode) derives the per-entity Updates // from currentβ†’updated. Registered on the OWNER's reduce config; a // lightweight subscriber may not have it, so this can be null. @@ -737,7 +802,7 @@ private ChangeItem BuildChangeItem(TStream? current, TStream updated) // (ToDataChangeRequest) gets a well-formed Update. Without this the // change shipped as a Full with empty Updates and the write-back's // `Updates.Any()` filter dropped it, so the write never persisted. - var typeRegistry = Hub.ServiceProvider.GetService(); + var typeRegistry = hub.ServiceProvider.GetService(); if (typeRegistry is not null && typeRegistry.TryGetCollectionName(typeof(TStream), out var collection) && !string.IsNullOrEmpty(collection)) @@ -763,15 +828,15 @@ private ChangeItem BuildChangeItem(TStream? current, TStream updated) /// clobbered by an older Full. See /// . /// - private ChangeItem BuildFullChangeItem(TStream? current, TStream updated) + private ChangeItem BuildFullChangeItem(IMessageHub hub, TStream? current, TStream updated) { // ChangedBy = ClientId always (never empty; matches the echo-suppression filters, // never the per-instance StreamId). AccessContext is orthogonal. See BuildChangeItem. var changedBy = ClientId; // 🚨 ONLY the owning hub sets Version. Subscriber carries the base it read. - var version = OwnerVersion(); + var version = OwnerVersion(hub); - var typeRegistry = Hub.ServiceProvider.GetService(); + var typeRegistry = hub.ServiceProvider.GetService(); if (typeRegistry is not null && typeRegistry.TryGetCollectionName(typeof(TStream), out var collection) && !string.IsNullOrEmpty(collection)) @@ -951,10 +1016,23 @@ public void OnError(Exception error) // stream whose Hub was never bound β€” but it must be LOUD: the silent version cost // #2625 two months of an unreproducible CI flake, because everything downstream // simply waited forever with nothing logged. - if (Hub is not null) + if (Hub is { } errorHub) { - Hub.FailStartup(error); - Hub.OpenGate(SynchronizationGate); + errorHub.FailStartup(error); + errorHub.OpenGate(SynchronizationGate); + } + else if (Volatile.Read(ref hubReleased) == 1) + { + // 🚨 A SECOND way to reach the branch above, introduced by #3321 step 3, and it is + // benign β€” so it must NOT borrow that branch's LogError. The hub was RELEASED + // because it died (this stream is undisposed, but its hosted sub-hub was torn down + // by its parent), which means there is no Started task left to settle: nothing is + // waiting on it, because the hub that owned it is already Dead. Levelling this at + // Error would ship a stack trace to Loki for every ordinary circuit teardown. + logger.LogDebug(error, + "[SYNC_STREAM] OnError for {StreamId} (Reference={Reference}, Owner={Owner}) " + + "after its hub was released β€” the sub-hub is already gone, so there is no " + + "startup left to fault.", StreamId, Reference, Owner); } else { @@ -1009,7 +1087,10 @@ private static bool IsTransientHubTimeout(Exception? error) /// The disposable to dispose with the stream. public void RegisterForDisposal(IDisposable disposable) { - if (isDisposed || Hub is null) + // ONE read of the field, used everywhere below β€” a Dispose() on another thread can null it + // between the guard and the use (#3321 step 3). + var hub = Hub; + if (isDisposed || hub is null) { // Disposed stream β€” no hub to register on. Dispose the registrant // immediately so the caller doesn't leak it. The caller's intent @@ -1023,7 +1104,7 @@ public void RegisterForDisposal(IDisposable disposable) // without Dispose() ever being called on the stream. Everything else rides the composite, // which Dispose() walks SYNCHRONOUSLY (see the field note). if (Interlocked.Exchange(ref hubDisposalHooked, 1) == 0) - Hub.RegisterForDisposal(streamDisposables); + hub.RegisterForDisposal(streamDisposables); // CompositeDisposable.Add disposes the registrant immediately if the composite is already // disposed, so a registration racing Dispose() cannot leak. @@ -1037,12 +1118,14 @@ public void RegisterForDisposal(IDisposable disposable) /// The processed delivery, or a failed delivery if the stream is dead/disposed. public IMessageDelivery DeliverMessage(IMessageDelivery delivery) { - if (isDisposed || Hub is null) + // ONE read of the field β€” see RegisterForDisposal (#3321 step 3). + var hub = Hub; + if (isDisposed || hub is null) { logger.LogDebug("[SYNC_STREAM] DeliverMessage skipped for {StreamId} β€” stream is dead/disposed", StreamId); return delivery.Failed("Stream is disposed"); } - return Hub.DeliverMessage(delivery.ForwardTo(Hub.Address)); + return hub.DeliverMessage(delivery.ForwardTo(hub.Address)); } @@ -1055,7 +1138,9 @@ public void OnNext(ChangeItem value) { // A DISPOSED stream has no hub to post to; drop the value rather than // NRE'ing on Hub.Post. (Subscribers already saw Store.OnCompleted.) - if (isDisposed || Hub is null) + // ONE read of the field β€” see RegisterForDisposal (#3321 step 3). + var hub = Hub; + if (isDisposed || hub is null) { logger.LogDebug("[SYNC_STREAM] OnNext skipped for {StreamId} β€” stream is dead/disposed", StreamId); return; @@ -1074,7 +1159,7 @@ public void OnNext(ChangeItem value) // delivery.AccessContext naturally. No ImpersonateAsHub stamping // here β€” hub addresses were polluting CreatedBy on user-driven // writes via the AsyncLocal leak (fixed 2026-05-22). - Hub.Post(new SetCurrentRequest(value)); + hub.Post(new SetCurrentRequest(value)); } catch (Exception ex) { @@ -1217,6 +1302,29 @@ public SynchronizationStream( // that outlives it is exactly the leaked callback the quiescing budget flags. RegisterForDisposal(resyncSubscription); + // 🚨 THE HOOK THAT ACTUALLY RECLAIMS THE LEAK (#3321 step 3). A sync sub-hub is a HOSTED + // hub: its parent disposes it during the parent's own teardown β€” a Blazor circuit ending, a + // DisposeRequest, a recycle β€” while this stream is owned by a workspace somewhere else and + // is never told. Measured on a 26 h production replica: 1 496 hubs at RunLevel=Dead, of + // which only 11 sat under a stream that had itself been disposed. So Dispose() clearing the + // field would have reclaimed 11 of 1 496; this hook is what reaches the other 1 485. + // + // Registered AFTER the line above on purpose: that call is what hooks `streamDisposables` + // onto the hub (RegisterForDisposal's one-shot `hubDisposalHooked`), and the hub's + // composite disposes in registration order β€” so the stream's own registrants run FIRST and + // still see a bound Hub, and this release runs after them. + syncHub.RegisterForDisposal(_ => ReleaseHub()); + + // A hub that was already past the point of accepting registrants disposes this one inline, + // so the release may ALREADY have happened. A stream with no hub is not a stream: refuse + // exactly as the null branch above does, rather than handing back one that can never work. + if (Hub is null) + { + isDisposed = true; + Store.OnCompleted(); + throw new HubDisposingException(Host.Address, Reference); + } + // 🚨 Capture the creating user's identity ONCE, here on the thread that constructs // the stream β€” in production that is the circuit thread (cache.GetMeshNodeStream) // or the owner's SubscribeRequest handler, where AccessService.Context is the real @@ -1466,7 +1574,7 @@ d.Message is SetCurrentRequest or DeliveryFailure or GetDataResponse // (far-higher) version made the monotonicity guard drop the lower-versioned // render Fulls as stale, so a late layout-area subscriber stayed stuck on // "Building layout…" and never emitted its content. - SetCurrent(hub, new ChangeItem(value, StreamId, OwnerVersion())); + SetCurrent(hub, new ChangeItem(value, StreamId, OwnerVersion(hub))); observer.OnNext(System.Reactive.Unit.Default); }, observer.OnError, @@ -1484,7 +1592,7 @@ d.Message is SetCurrentRequest or DeliveryFailure or GetDataResponse { // Same one-clock invariant as the observable-init path above: OwnerVersion(), // never Host.Version, so the init frame can't outrank later owned writes. - SetCurrent(hub, new ChangeItem(init, StreamId, OwnerVersion())); + SetCurrent(hub, new ChangeItem(init, StreamId, OwnerVersion(hub))); return System.Reactive.Unit.Default; }) // 🚨 A faulted initial load must fault the STREAM, not only this hub's buildup. @@ -1513,7 +1621,7 @@ private void UpdateStream(IMessageDelivery delivery, IMessageH logger.LogDebug("[SYNC_STREAM] UpdateStream called for {StreamId}, ChangeType={ChangeType}, Version={Version}, MessageId={MessageId}", StreamId, delivery.Message.ChangeType, delivery.Message.Version, delivery.Id); - if (Hub is null || Hub.IsDisposing) + if (Hub is not { IsDisposing: false }) { logger.LogDebug("[SYNC_STREAM] UpdateStream skipped for {StreamId} - hub is disposing/dead", StreamId); return; @@ -2246,8 +2354,16 @@ public void Dispose() "[SYNC_STREAM] Registrant faulted while disposing stream {StreamId}", StreamId); } - if (Hub is not null && Hub.RunLevel <= MessageHubRunLevel.Started) - Hub.Dispose(); + if (Hub is { RunLevel: <= MessageHubRunLevel.Started } ownHub) + ownHub.Dispose(); + + // 🚨 DROP THE REFERENCE β€” the whole point of #3321 step 3. Hub.Dispose() above only STARTS + // the hub's teardown (it posts ShutdownRequest and returns), and closing the hub's Autofac + // scope β€” which HostedHubsCollection does, correctly, several turns later β€” does not null + // the fields of an object that outlives it. Until this line the stream kept the corpse and + // everything it had resolved (its own ILifetimeScope, its own TypeRegistry) reachable + // forever. The hub-death hook registered in the constructor covers the other direction. + ReleaseHub(); } private ConcurrentDictionary Properties { get; } = new(); /// Reads a property bag value by key. diff --git a/src/MeshWeaver.Data/StandardReducers.cs b/src/MeshWeaver.Data/StandardReducers.cs index 5cc0580d74..ed72a5d32f 100644 --- a/src/MeshWeaver.Data/StandardReducers.cs +++ b/src/MeshWeaver.Data/StandardReducers.cs @@ -106,21 +106,31 @@ private static ChangeItem ReduceJsonElementTo(ChangeItem`: it MUST return a change, so there is no absent value to hand back and no + // error channel but a throw. Since step 3 a disposed stream RELEASES its hub, so `stream.Hub` + // can be absent while the reducer runs (the transform executes on the sync hub's turn, which a + // Dispose() on another thread can overtake); reading the field four times, as the code below + // used to, is four chances to NRE. RequireHub() reads it once and refuses with the TRANSIENT + // HubDisposingException, which the single gateway (JsonSynchronizationStream.ToChangeItem) + // turns into the `null` its own signature already models β€” "no patch derived", not a fault. private static ChangeItem PatchJsonElement(ISynchronizationStream stream, JsonElement current, JsonElement updated, JsonPatch? patch, string changedBy) { - var typeRegistry = stream.Hub.TypeRegistry; - return new(updated, changedBy, stream.StreamId, ChangeType.Patch, stream.Current?.Version ?? stream.Hub.Version, current.ToEntityUpdates(updated, patch!, stream.Hub.JsonSerializerOptions, typeRegistry)); + var hub = stream.RequireHub(); + var typeRegistry = hub.TypeRegistry; + return new(updated, changedBy, stream.StreamId, ChangeType.Patch, stream.Current?.Version ?? hub.Version, current.ToEntityUpdates(updated, patch!, hub.JsonSerializerOptions, typeRegistry)); } private static ChangeItem PatchInstanceCollectionJsonElement(ISynchronizationStream stream, InstanceCollection current, JsonElement updated, JsonPatch? patch, string changedBy) { - var updatedInstances = updated.Deserialize(stream.Hub.JsonSerializerOptions)!; + var hub = stream.RequireHub(); + var updatedInstances = updated.Deserialize(hub.JsonSerializerOptions)!; return new( updatedInstances, changedBy, stream.StreamId, ChangeType.Patch, - stream.Current?.Version ?? stream.Hub.Version, - current.ToEntityUpdates((CollectionReference)stream.Reference, updated, patch!, stream.Hub.JsonSerializerOptions)); + stream.Current?.Version ?? hub.Version, + current.ToEntityUpdates((CollectionReference)stream.Reference, updated, patch!, hub.JsonSerializerOptions)); } private static ChangeItem ReduceInstanceCollectionTo(ChangeItem current, InstanceReference reference, bool initial) @@ -266,6 +276,10 @@ private static ChangeItem ReduceEntityStoreTo(ChangeItem cu private static ChangeItem PatchEntityStore(ISynchronizationStream stream, EntityStore currentStore, JsonElement updatedJson, JsonPatch? patch, string changedBy) { + // One hub read for the whole reduction, threaded into the two helpers below rather than + // re-read at each of the four sites this method used to touch it β€” see the note above + // (#3321 step 3). Reading it four times is four chances to catch a concurrent release. + var hub = stream.RequireHub(); var updates = new List(); foreach (var g in patch!.Operations @@ -279,7 +293,7 @@ private static ChangeItem PatchEntityStore(ISynchronizationStream new EntityUpdate(collection, x.Key, x.Value))); @@ -309,7 +323,7 @@ private static ChangeItem PatchEntityStore(ISynchronizationStream(eg.Key.Id, stream.Hub.JsonSerializerOptions)!; + var id = JsonSerializer.Deserialize(eg.Key.Id, hub.JsonSerializerOptions)!; var currentCollection = currentStore.GetCollection(collection); if (currentCollection == null) throw new InvalidOperationException( @@ -320,12 +334,12 @@ private static ChangeItem PatchEntityStore(ISynchronizationStream PatchEntityStore(ISynchronizationStream stream, JsonPointer entityPointer, JsonElement updatedJson) + // Both take the RESOLVED hub rather than the stream: PatchEntityStore reads it once and + // threads it through, so there is exactly one point at which a released stream can be found + // (#3321 step 3). + private static object GetEntity(IMessageHub hub, JsonPointer entityPointer, JsonElement updatedJson) { var entity = entityPointer.Evaluate(updatedJson); - return entity?.Deserialize(stream.Hub.JsonSerializerOptions)!; + return entity?.Deserialize(hub.JsonSerializerOptions)!; } - private static InstanceCollection DeserializeCollection(this ISynchronizationStream stream, JsonElement updatedJson, JsonPointer pointer) + private static InstanceCollection DeserializeCollection(IMessageHub hub, JsonElement updatedJson, JsonPointer pointer) { - return pointer.Evaluate(updatedJson)!.Value.Deserialize(stream.Hub.JsonSerializerOptions)!; + return pointer.Evaluate(updatedJson)!.Value.Deserialize(hub.JsonSerializerOptions)!; } /// diff --git a/src/MeshWeaver.Data/SynchronizationStreamLiveness.cs b/src/MeshWeaver.Data/SynchronizationStreamLiveness.cs index be066babe0..bc065d920b 100644 --- a/src/MeshWeaver.Data/SynchronizationStreamLiveness.cs +++ b/src/MeshWeaver.Data/SynchronizationStreamLiveness.cs @@ -34,11 +34,14 @@ namespace MeshWeaver.Data; /// JsonSynchronizationStream and StreamNotConvergingException already point readers at /// "StreamLiveness.IsUsable", and a consumer following one of them must find the thing it names. /// -/// 🚨 This is step 1 alone, and it changes no behaviour. Steps 2 and 3 of #3321 β€” -/// migrating the 47 dereference sites onto , and only THEN dropping the -/// Hub reference on disposal so the leaked hub graph is released β€” are deliberately NOT done -/// here. Dropping the reference before the call sites can answer "no" is precisely the production -/// incident above, in the same order it happened. +/// 🚨 All three steps of #3321 have landed, in the only order that works. Step 1 +/// (#3380) added these two methods and changed no behaviour. Step 2 (#3386) migrated the consumer +/// dereference sites onto . Step 3 then released the reference β€” from BOTH +/// ends, because only 11 of the 1 496 corpses in the production dump sat under a stream that had +/// itself been disposed; the other 1 485 were hosted sub-hubs killed by their parent's teardown +/// under a stream nobody ever disposed. Dropping the reference before the call sites could answer +/// "no" is precisely the production incident above, in the same order it happened β€” which is why +/// this file came first. See Doc/Architecture/StreamLivenessAndTheHubReference. /// public static class SynchronizationStreamLiveness { @@ -78,4 +81,71 @@ public static bool IsUsable(this ISynchronizationStream? stream) /// mirrors something that is. public static IMessageHub? TryGetHub(this ISynchronizationStream? stream) => stream.IsUsable() ? stream!.Hub : null; + + /// + /// The stream's hub if it still HOLDS one, otherwise null β€” the PRESENCE accessor, and + /// the migration target for an OWNER-SIDE stream.Hub.Something whose previous behaviour + /// was unconditional (Systemorph/MeshWeaver#3321, step 3 of 3). + /// + /// 🚨 Presence is a different question from liveness, and using the wrong one is a + /// behaviour change β€” measured, not theorised. Migrating + /// DataSource.Initialized (Task.WhenAll(streams.Select(s => s.Hub.Started))) onto + /// silently BROKE + /// DataContextFaultedInitBeforeStreamHubBoundTest: a stream whose initial load faults is + /// not β€” that is the whole point of IsFaulted β€” so the guard + /// excluded it from the WhenAll, Initialized completed SUCCESSFULLY, and a hub whose + /// data source had thrown started answering requests as though nothing had happened. Faulting + /// that Started task is exactly what must still be awaited. + /// + /// So the rule is: where the site is asking "should I use this + /// stream?" (a consumer, a cache, a read); this one where the site is asking "is the reference + /// still there?" (an owner writing into its own stream, where refusing a merely winding-down + /// hub would change what the code does). Step 2 declined to tighten already-working sites + /// for the same reason, and this method is what lets step 3 make them null-safe without + /// tightening them either. + /// + /// Answers null for a null stream, so a nullable stream can be tested + /// directly. It reads the field ONCE β€” which is the other half of the point: a caller that + /// re-reads stream.Hub after its own guard is check-then-act across threads, because a + /// Dispose() elsewhere can land in between. + /// + /// The stream to read the hub from, or null. + /// The hub, or null when the stream is null or has released it. + public static IMessageHub? HubIfHeld(this ISynchronizationStream? stream) + => stream?.Hub; + + /// + /// The stream's hub, or a TRANSIENT when the stream has + /// RELEASED it β€” the same PRESENCE question as , for a surface whose + /// signature admits no absent value (Systemorph/MeshWeaver#3321, step 3 of 3). + /// + /// 🚨 This is NOT a liveness predicate and must never grow into one. + /// is the ONE liveness answer β€” it walks the reduce chain, counts a + /// faulted store, and refuses a hub that has merely begun winding down. This method asks a + /// strictly narrower and purely factual question: is the reference still there? Step 3 + /// made Hub clearable β€” SynchronizationStream.Dispose() and the stream's + /// hub-death hook both drop it, which is what releases the leaked + /// stream β†’ dead hub β†’ resolved state graph β€” so "absent" became a state the field can + /// really be in. Tightening this to would refuse a WINDING-DOWN or + /// FAULTED but still-present hub and so change behaviour on live paths (see + /// for the measured instance); keeping them separate is also what stops + /// this from becoming the fourth hand-copied liveness predicate (#1455). + /// + /// Why a throw is the right refusal here. The callers are patch reducers + /// (StandardReducers, MeshDataSource.PatchMeshNode) whose signature is + /// … β†’ ChangeItem<T>: they MUST return a change, so a throw is the only channel + /// they have, and inventing a sentinel change would be worse than saying so. + /// is an , so + /// it classifies as ErrorType.ShuttingDown β€” the transient "ask again" answer β€” instead + /// of the terminal DeliveryFailure the predecessor's raw NRE produced. Their single + /// gateway, JsonSynchronizationStream.ToChangeItem, catches exactly this type and + /// answers with the null its own signature already models, so a reducer racing a + /// disposal degrades to "no patch derived" rather than to a fault. + /// + /// The stream to read the hub from. + /// The stream's hub; never null. + /// The stream has released its hub. + public static IMessageHub RequireHub(this ISynchronizationStream stream) + => stream.HubIfHeld() + ?? throw new HubDisposingException(stream.Host.Address, stream.Reference); } diff --git a/src/MeshWeaver.Data/VirtualDataSource.cs b/src/MeshWeaver.Data/VirtualDataSource.cs index e58e8267e6..7c9f2b11d2 100644 --- a/src/MeshWeaver.Data/VirtualDataSource.cs +++ b/src/MeshWeaver.Data/VirtualDataSource.cs @@ -171,10 +171,18 @@ protected override ISynchronizationStream SetupDataSourceStream( using (accessService?.ImpersonateAsSystem()) stream.Update(store => { + // 🚨 #3321 step 3: this transform runs on the sync hub's turn, which + // a Dispose() on another thread can overtake β€” and a released stream + // has no hub to stamp a Version from. `null` is this delegate's own + // documented no-op, so the refusal costs nothing and invents nothing. + // PRESENCE, not liveness: it mirrors the `TryGetActiveHub` guard + // Update already applied before posting this transform. + if (stream.HubIfHeld() is not { } streamHub) + return null; var newStore = (store ?? new EntityStore()) .WithCollection(typeSource.CollectionName, collection); return (ChangeItem?) - new ChangeItem(newStore, Id.ToString()!, stream.StreamId, ChangeType.Full, stream.Hub.Version, []); + new ChangeItem(newStore, Id.ToString()!, stream.StreamId, ChangeType.Full, streamHub.Version, []); }, _ => { }); }, // 🚨 THE ERROR ARM IS NOT OPTIONAL β€” omitting it is a PROCESS KILLER, not a diff --git a/src/MeshWeaver.Data/Workspace.cs b/src/MeshWeaver.Data/Workspace.cs index c801b917a5..f3b5942478 100644 --- a/src/MeshWeaver.Data/Workspace.cs +++ b/src/MeshWeaver.Data/Workspace.cs @@ -1339,7 +1339,18 @@ internal int EvictClientSubscriptions(string subscriberPath) // sync hub (SynchronizationStream assigns its own sync/{clientId} sub-hub to // .Hub β€” never the owner hub). The stream's own disposal registration then finds // the entry already gone and does nothing. - kv.Value.Stream.Hub.Dispose(); + // 🚨 A NULL check, deliberately NOT a liveness check (#3321 step 3). Guarding this + // with IsUsable would SKIP a disposal β€” a reduced stream whose PARENT died first is + // unusable while its own hub is still very much alive, and leaving that one + // undisposed is the leak this whole issue is about. `Hub is { }` only skips when + // the stream has already RELEASED the reference, which since step 3 means the hub + // is disposed: there is nothing left to dispose and nothing left to leak. + if (kv.Value.Stream.Hub is { } streamHub) + streamHub.Dispose(); + else + _logger.LogDebug( + "Workspace {WorkspaceId}: server-side stream {StreamId} for {Subscriber} had " + + "already released its hub β€” nothing to dispose", Id, kv.Key.StreamId, subscriberPath); evicted++; } catch (Exception ex) diff --git a/src/MeshWeaver.Data/WorkspaceExtensions.cs b/src/MeshWeaver.Data/WorkspaceExtensions.cs index d5047e8634..7b117575ff 100644 --- a/src/MeshWeaver.Data/WorkspaceExtensions.cs +++ b/src/MeshWeaver.Data/WorkspaceExtensions.cs @@ -92,6 +92,11 @@ public static IWorkspace GetWorkspace(this IMessageHub messageHub) => /// The stream the change is attributed to (provides id and version). /// The new store together with the list of entity updates that produced it. /// A patch change item ready to be pushed into the stream. + /// The stream has released its hub + /// (#3321 step 3). Like a patch reducer, this helper MUST return a change, so a throw is its + /// only refusal channel β€” and this one is transient (ErrorType.ShuttingDown), so a caller + /// inside a stream.Update(…) lambda routes it to its own exceptionCallback + /// instead of NRE'ing on a released field. public static ChangeItem ApplyChanges( this ISynchronizationStream? stream, EntityStoreAndUpdates storeAndUpdates) => @@ -99,7 +104,7 @@ public static ChangeItem ApplyChanges( storeAndUpdates.ChangedBy ?? stream!.StreamId, stream!.StreamId, ChangeType.Patch, - stream!.Hub.Version, + stream!.RequireHub().Version, storeAndUpdates.Updates.ToArray() ); diff --git a/src/MeshWeaver.Data/WorkspaceStreams.cs b/src/MeshWeaver.Data/WorkspaceStreams.cs index 4873c112ac..9e42c67c1e 100644 --- a/src/MeshWeaver.Data/WorkspaceStreams.cs +++ b/src/MeshWeaver.Data/WorkspaceStreams.cs @@ -84,6 +84,21 @@ public static class WorkspaceStreams { try { + // 🚨 #3321 step 3: the stream is the SUBSCRIBER here, and a subscription callback + // can be mid-flight when the stream is disposed or its hub torn down β€” at which + // point `ret.Hub` is released. Skipping the whole emission is the "no" this + // callback already models: it returns void, and `ret.OnNext` on a released stream + // is a documented drop, so the frame had nowhere to land either way. PRESENCE, not + // liveness (HubIfHeld, not TryGetHub) β€” this guard mirrors OnNext's own + // `isDisposed || Hub is null`, so it can never refuse a frame that would still + // have landed. Resolved ONCE and reused for both stamps below, never re-read. + if (ret.HubIfHeld() is not { } retHub) + { + logger.LogDebug( + "Combined stream {StreamId} for {Address} is torn down β€” dropping this frame", + ret.StreamId, workspace.Hub.Address); + return; + } logger.LogDebug("Retrieved values {Changes}", changes.Select(c => JsonSerializer.Serialize(c, ret.Host.JsonSerializerOptions))); if (!isInitialized) { @@ -92,7 +107,7 @@ public static class WorkspaceStreams .Where(c => c.Value != null) .Aggregate(new EntityStore(), (acc, change) => acc.Merge(change.Value!)); - var initialChange = new ChangeItem(initialStore, ret.StreamId, ret.Hub.Version); + var initialChange = new ChangeItem(initialStore, ret.StreamId, retHub.Version); ret.OnNext(initialChange); @@ -115,7 +130,7 @@ public static class WorkspaceStreams .Where(c => c.Value != null) .Aggregate(new EntityStore(), (acc, change) => acc.Merge(change.Value!)); - var updateChange = new ChangeItem(updatedStore, ret.StreamId, ret.Hub.Version) + var updateChange = new ChangeItem(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(), diff --git a/src/MeshWeaver.Documentation/Data/Architecture/StreamLivenessAndTheHubReference.md b/src/MeshWeaver.Documentation/Data/Architecture/StreamLivenessAndTheHubReference.md index b90e51d88e..fa61fb9b82 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/StreamLivenessAndTheHubReference.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/StreamLivenessAndTheHubReference.md @@ -1,21 +1,25 @@ --- Name: Stream Liveness and the Hub Reference Category: Architecture -Description: A synchronization stream outlives the hub it holds, and ISynchronizationStream.Hub is declared non-nullable β€” so a corpse answers it and 47 call sites dereference it. Why the obvious fix (null the field) is the one that already caused a production NRE, and the three-step order that does not repeat it. +Description: A synchronization stream outlived the hub it held, and ISynchronizationStream.Hub is declared non-nullable β€” so a corpse answered it and 54 call sites dereferenced it. Why the obvious fix (null the field) is the one that already caused a production NRE, and the three-step order that did not repeat it. All three steps have landed. Icon: --- # Stream Liveness and the Hub Reference **A `SynchronizationStream` holds `IMessageHub Hub`, the interface declares it NON-NULLABLE, and a -stream whose owner has been torn down keeps answering it. So the corpse is indistinguishable from a -live stream at every one of the 47 sites that dereference it β€” and the reference keeps the dead -hub's whole resolved graph alive.** +stream whose owner had been torn down kept answering it. So the corpse was indistinguishable from a +live stream at every one of the 54 sites that dereference it β€” and the reference kept the dead hub's +whole resolved graph alive.** That is two defects wearing one shape: a **retention** leak, and a **contract** that lies. They have to be fixed in that order backwards β€” the contract first β€” and this page exists because the obvious order was already tried in production. +**All three steps have landed** (#3380, #3386, #3321). The stream now releases the reference from +both ends β€” its own disposal, and its hub's death underneath it β€” and the sections below say why +that could not simply be done, and what each of the 54 sites answers now. + ## What the field costs while it is held Closing a DI scope does not null the fields on an object that outlives it. The retained graph is @@ -27,12 +31,13 @@ hubs were removed. The bytes stay reachable anyway, because reachability is abou scopes. The only way to reclaim them is to **drop the reference**. `Dispose()` sets `isDisposed`, completes the store, clears `sharedReduceCache` and disposes -`streamDisposables` β€” but it never clears `Hub`. So even the streams that *are* properly disposed -keep their hub reachable. +`streamDisposables` β€” but it never cleared `Hub`. So even the streams that *were* properly disposed +kept their hub reachable. 🚨 And that is the smaller half: only **11** of the 1 496 corpses sat under +a disposed stream. See "What step 3 did" below. -## Why "just null the field" is not available +## Why "just null the field" was never available on its own -It was already done, and it is recorded in `SynchronizationStream`'s own constructor refusal path. +It was already done once, and it is recorded in `SynchronizationStream`'s own constructor refusal path. The predecessor built a DEAD stream β€” `isDisposed`, store completed, **`Hub = null!`** β€” and documented that *"every code path that touches Hub goes through `TryGetActiveHub`"*. @@ -68,10 +73,12 @@ that, letting a consumer **ask**. ## The three steps, in the only order that works 1. βœ… **Expose the predicate.** `SynchronizationStreamLiveness.IsUsable(stream)` and - `stream.TryGetHub()` make the existing answer reachable from other assemblies. + `stream.TryGetHub()` make the existing answer reachable from other assemblies. (Step 3 adds + `HubIfHeld()` and `RequireHub()` beside them β€” the PRESENCE question, which is not the same + question; see below.) 2. βœ… **Migrate the dereference sites** onto `TryGetHub()`, which can answer `null`. -3. **Only then** clear `Hub` on disposal β€” at which point a null hub is a state the contract admits - rather than a lie. **Not done.** +3. βœ… **Only then** release `Hub` β€” at which point an absent hub is a state the contract admits + rather than a lie. Doing 3 before 2 is the production NRE, in the order it happened. Doing 2 before 1 is impossible. @@ -120,13 +127,213 @@ The 26 left fall in three groups, and the reason differs per group: inventing a failure mode rather than migrating onto an existing one β€” so they are named here instead. **They are step 3's remaining work**: before `Hub` can be cleared on disposal, each of these needs a decision about what a reducer does when its stream has died, and that is a design - question, not a mechanical substitution. + question, not a mechanical substitution. (Step 3 resolved them β€” plus five more this list did not + cover. See "How the sites step 2 could not migrate were resolved" below.) One more constraint the migration follows: **a guarded read inside a per-emission lambda re-reads `TryGetHub()` rather than capturing the hub resolved at pipeline-construction time.** Capturing would pin the hub's whole resolved graph for the lifetime of every subscription β€” the exact retention this issue exists to release. +## What step 3 did + +### 🚨 `Dispose()` alone would have reclaimed 11 of 1 496 + +The step is usually described as "`SynchronizationStream.Dispose()` clears `Hub`". Read against the +dump, that description is a **sixth of the fix**: + +``` +SynchronizationStream total=8461 disposed=11 + 1485 hub RunLevel=Dead, stream NOT disposed + 11 hub RunLevel=Dead, stream DISPOSED +``` + +A `sync/` hub is a HOSTED hub. Its parent disposes it during the parent's own teardown β€” a Blazor +circuit ending, a `DisposeRequest`, a recycle β€” while the stream that created it is owned by a +workspace somewhere else and **is never told**. So the stream keeps `isDisposed == false`, keeps a +strong reference to the corpse, and keeps being handed out as usable. Clearing the field in +`Dispose()` never runs for those 1 485. + +Step 3 therefore releases from **both ends**, through one private `ReleaseHub()` (idempotent, CAS'd, +callable from any thread β€” it disposes nothing, it only drops a reference): + +| End | Where | Covers | +|---|---|---| +| The stream is disposed | `Dispose()`, after `Hub.Dispose()` | the 11 | +| The hub dies underneath it | `syncHub.RegisterForDisposal(_ => ReleaseHub())`, from the constructor | the 1 485 | + +The constructor's hook is registered **after** `RegisterForDisposal(resyncSubscription)`, and that +ordering is load-bearing: that call is what hooks the stream's own `streamDisposables` onto the hub +(`RegisterForDisposal`'s one-shot `hubDisposalHooked`), and a hub's composite disposes in +registration order β€” so the stream's registrants run first and still see a bound `Hub`, and the +release runs after them. A hub already past accepting registrants disposes the hook inline, so the +constructor re-checks and refuses with `HubDisposingException` exactly as its `syncHub is null` +branch does: a stream with no hub is not a stream. + +Nothing new appears in the liveness vocabulary. `StreamLiveness.IsUsable` has always treated +`current.Hub is not { } hub` as dead, so a released stream reports itself unusable through the same +predicate every cache already consults. + +### How the sites step 2 could not migrate were resolved + +A fresh sweep of `src/` β€” every `.Hub` whose receiver resolves to an `ISynchronizationStream` β€” +found **29 occurrences on 27 lines**. Four stay untouched and must: two ARE the predicate +(`StreamLiveness.IsUsable`'s chain walk, `TryGetHub`'s own read), one is already written `stream?.Hub?.…`, +and one is `Workspace`'s cache probe, which is itself a liveness check. **The other 25 are step 3's +work** β€” the 20 step 2 named, plus five it did not: + +- the **three** `LayoutAreaHost` reads step 2 classified "hub-turn confined". That classification + rested on *a hub executing a turn is by construction alive*, which is a statement about the HUB. + Step 3 makes the FIELD independently clearable, so a queued `UpdateStreamRequest` can now run + after the release β€” the premise expired with the change that needed it; +- `Workspace.EvictClientSubscriptions`, which step 2 placed in "already answerable" for the right + reason (guarding it with `IsUsable` would skip a disposal) β€” a null check is still needed; +- `WorkspaceExtensions.ApplyChanges`, which step 2's list **missed** (`stream!.Hub.Version`, behind + a null-forgiving `!`). + +The principle is unchanged from step 2: **answer with the "no" the surface already models; where +there is none, use the channel it does have.** + +| Group | Sites | Resolution | +|---|---|---| +| **Patch reducers** β€” `StandardReducers` (10), `MeshDataSource.PatchMeshNode` (2) | 12 | `RequireHub()`, absorbed by their single gateway | +| **A creation-path `Subscribe` callback** β€” `WorkspaceStreams` (2), `PartitionedHubDataSource` (2) | 4 | `HubIfHeld()` once at the top; **skip the emission** | +| **Inside a `stream.Update(…)` lambda** β€” `LayoutAreaHost` (3), `VirtualDataSource` (1) | 4 | `HubIfHeld()`, then `return null` β€” `Update`'s documented no-op | +| `JsonSynchronizationStream`'s `reduced.Hub.Register` | 2 | `HubIfHeld()` once; **don't register** | +| `GenericUnpartitionedDataSource.Initialized` | 1 | a released stream contributes **no task to wait for** | +| `WorkspaceExtensions.ApplyChanges` | 1 | `RequireHub()` β€” same shape as a reducer | +| `Workspace.EvictClientSubscriptions` | 1 | `Hub is { }` β€” deliberately not a liveness check | +| | **25** | | + +**The reducers are the whole design problem, and the gateway is the answer.** A reducer's signature +is `… β†’ ChangeItem`: it MUST return a change, so it has no absent value to hand back. But every +patch reducer in the codebase is invoked from exactly ONE place β€” +`JsonSynchronizationStream.ToChangeItem`, whose `PatchFunction?.Invoke` is the only call site of +`ReduceManager.PatchFunction` β€” and `null` is what `ToChangeItem` **already returns** for a stream +with no `PatchFunction` registered. So: + +- the reducers read the hub once through `SynchronizationStreamLiveness.RequireHub()`, which refuses + with the TRANSIENT `HubDisposingException` (an `ObjectDisposedException`, classified + `ErrorType.ShuttingDown` β€” the "ask again" answer, never the terminal `DeliveryFailure` the + predecessor's raw NRE produced); +- `ToChangeItem` catches **only that type** and answers `null`, which both of its callers already + handle: `BuildChangeItem` falls through to its type-registry path, `UpdateStream` falls back to + deserializing the patched JSON through `Host.JsonSerializerOptions`. + +That is not a swallow, and it is not a `catch (NullReferenceException)`. It converts one specific, +meaningful refusal into an absence the signature already carries; every other exception a reducer +throws still propagates. + +### 🚨 PRESENCE is not LIVENESS, and using the wrong one is a behaviour change β€” measured + +The first draft of this step wrote every owner-side guard as `TryGetHub()`, on the reasoning that +step 2's rule was "migrate onto `TryGetHub()`". That is wrong, and one existing test proved it before +the change left the branch. + +`DataSource.Initialized` is `Task.WhenAll(streams.Select(s => s.Hub.Started))`. Guarded with +`TryGetHub()`, a stream whose **initial load faulted** is excluded β€” because it is not `IsUsable`, +which is exactly what `IsFaulted` is for. So the WhenAll no longer contained the task carrying the +fault, `Initialized` completed **successfully**, and a hub whose data source had thrown went on +answering requests as though nothing had happened. `DataContextFaultedInitBeforeStreamHubBoundTest` +went red on precisely that (*"a hub whose data-source init threw must answer requests with an error, +but no exception was thrown"*) β€” the #2625 wedge, re-introduced by a guard that looked like a +tightening of safety. + +So step 3 uses **two different accessors, for two different questions**: + +| Accessor | Question | Where | +|---|---|---| +| `TryGetHub()` (`IsUsable`) | *should I use this stream?* β€” walks the reduce chain, counts a faulted store, refuses a winding-down hub | a CONSUMER, a cache, a read β€” step 2's 28 sites | +| `HubIfHeld()` / `RequireHub()` | *is the reference still there?* β€” one field read, nothing else | an OWNER writing into its own stream, where a stricter answer changes what the code does | + +**The rule for an owner-side guard is: match the guard the site's own gate already applies.** Every +one of them ends in `stream.OnNext` or `stream.Update`, whose refusal is `isDisposed || Hub is null` +β€” so a presence check refuses in exactly the cases the write would have been dropped anyway, and +never one more. `TryGetHub()` there would be a silent policy change smuggled in under a null-safety +fix, which is the shape step 2 already declined at `Workspace`'s cache probe. + +`RequireHub()` and `HubIfHeld()` therefore ask the same narrow question and one is written in terms +of the other, so there is one definition of presence and one of liveness β€” never a fourth +hand-copied predicate (#1455). The same reasoning governs `Workspace.EvictClientSubscriptions`: +guarding its `Stream.Hub.Dispose()` with `IsUsable` would SKIP a disposal β€” a reduced stream whose +*parent* died first is unusable while its own hub is very much alive β€” so it is guarded with +`Hub is { }`, which only skips when the reference is already gone, i.e. when the hub is already +disposed. + +### 🚨 The dependent half β€” and why no gate could have told you + +`ISynchronizationStream.Hub` is a **public** member of a **public** interface that these assemblies +ship as packages, and step 3 changes what it ANSWERS without changing its signature. That is shape 7 +of the seven cross-repo break shapes, and it is the one **no surface detector can see by +construction** β€” the `Cross-repo pair (public surface)` gate resolves removed types and removed +members, and this diff removes zero of each. So the sweep has to be done by hand, on the receiver +type, across every satellite checkout. + +Measured on 2026-09-06, all six satellites, every file (not just `.cs` β€” in-mesh C# lives inside +`.json` node strings too): + +| Repo | stream-receiver `.Hub` | raw `.Hub` lines classified | +|---|---:|---:| +| **MeshWeaver.Plugins** | **22** (11 production, 11 test) | 2 051 | +| education | 0 | 483 | +| MeshWeaver.Education | 0 | 11 | +| MeshWeaver.Reinsurance | 0 | 152 | +| MeshWeaver.SocialMedia | 0 | 73 | +| MeshWeaver.Manufacturing | 0 | 15 | + +The four content satellites are clean *structurally*, not merely by token match: +`ISynchronizationStream` appears in ZERO files in four of them, and in exactly one file in +Reinsurance where the stream is only `.Update(…)`d. Every `.json` node carrying `.Hub` inside a C# +`configuration` string resolves to `IWorkspace.Hub` / `LayoutAreaHost.Hub` / an `IMessageHub` local. + +**All 11 production dereferences are one cluster in `MeshWeaver.Plugins`** β€” the Blazor view layer's +inherited `ISynchronizationStream? Stream` (`BlazorView.razor.cs`): `OnClick`, +`OnBlur`, both `DialogView` close handlers, and seven `JsonSerializerOptions` reads across +`ViewModelExtensions`, `FormComponentBase`, `DataGridView`, `RadzenChartView` and +`RadzenPivotGridView`. + +🚨 **The `!` in `Stream!.Hub.Something` is on `Stream`, not on `.Hub`** β€” so the NRE moves one level +right, past the operator that suppresses the warning, and the compiler stays silent. One of them +(`RadzenChartView.razor:171`) sits inside a bare `catch { }`, so it would have been invisible at +runtime too: a wrong chart rather than an exception. + +**Order: the dependent lands FIRST.** Its guards are no-ops on a live stream, so they are safe +against the old core; core's release is not safe against an unguarded dependent. The core PR was +held as a draft until the Plugins counterpart merged. + +### And every read inside the stream now resolves the field once + +Step 2 excluded `SynchronizationStream`'s own bare `Hub` reads as "already guarded". They were β€” +against a field that could not change. Once `Dispose()` can null it, a guard followed by a *re-read* +is check-then-act across threads, so every read below a guard now uses the local the guard resolved: +`OwnerVersion(hub)`, `BuildChangeItem(hub, …)`, `BuildFullChangeItem(hub, …)`, and the locals in +`RegisterForDisposal` / `DeliverMessage` / `OnNext` / `OnError`. `Update`'s and `SetFull`'s value +transforms resolve through `TryGetActiveHub` on the turn and return `null` when it answers no. + +### The measurement that says these tests could fail + +`StreamReleasesItsHubTest` (6 cases; the numbers below were taken at 5, before the accessor-parity +case was added) and the step-2 suite were re-run with **only** the two +`ReleaseHub()` call sites disabled β€” everything else, including `RequireHub` and every guard, left +in place, so the experiment isolates exactly what step 3 does: + +| Suite | release disabled | release enabled | +|---|---|---| +| `StreamReleasesItsHubTest` (`MeshWeaver.Data.Test`) | **4 failed, 1 passed** | **5 passed** | +| `TornDownStreamCallSitesTest` (`MeshWeaver.Layout.Test`) | **4 failed, 1 passed** | **5 passed** | + +The live control passed in both runs of both suites β€” a release that fired unconditionally, or a +guard that answered "dead" always, would satisfy every negative at once, and the controls are what +rule that out. + +🚨 The flagship assertion is a `WeakReference`, not `Hub is null`. "The field reads null" is a +statement about one field; **"the hub is unreachable"** is the statement this issue is about, and it +is the only one that goes red on the defect for the right reason. It passes: with the release in +place, the hub is collected. + +(A *full* `src/` revert does not compile the tests, because `RequireHub` is itself part of step 3 β€” +which is why the falsification reverts the release rather than the whole change.) + ### Why step 1 is an extension, not an interface member `StreamLiveness` records both reasons and neither has expired: diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-06-a-closed-page-gives-its-memory-back.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-06-a-closed-page-gives-its-memory-back.md new file mode 100644 index 0000000000..e0ee102ce2 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-06-a-closed-page-gives-its-memory-back.md @@ -0,0 +1,27 @@ +--- +Name: A closed page gives its memory back +Category: Fix +Description: Closing a page or ending a session used to leave part of it behind in the server's memory. Those leftovers are now released, so a portal that has been running for days stays as responsive as one that just started. +Icon: TopSpeed +Order: -20260906 +--- + +# A closed page gives its memory back + +Every open view β€” a document, a dashboard, a chat thread β€” is backed by a small piece of machinery on +the server that keeps it in sync with the data behind it. When you close the page, or your session +ends, or the portal recycles a node, that machinery is shut down. + +**Shut down, but until now not entirely let go.** Something still held a reference to it, so the +memory it used could never be reclaimed. On a busy portal that added up: a server running for a +couple of days accumulated thousands of these leftovers, and the garbage collector could do nothing +about them because, as far as it could tell, they were still in use. The symptom people actually +noticed was the portal getting steadily slower the longer it had been up β€” long pauses, then a +restart that made everything feel fast again for a day. + +The leftovers are now released, from both directions: when a view is closed normally, and when the +session behind it is torn down without the view being told. A portal that has been up for days now +behaves like one that started an hour ago, and the periodic slow-down-then-restart cycle goes with +it. + +Nothing about how views work has changed β€” this is purely what happens after they are finished with. diff --git a/src/MeshWeaver.Graph/MeshDataSource.cs b/src/MeshWeaver.Graph/MeshDataSource.cs index 02f25d652b..d64d70f6a7 100644 --- a/src/MeshWeaver.Graph/MeshDataSource.cs +++ b/src/MeshWeaver.Graph/MeshDataSource.cs @@ -1924,8 +1924,12 @@ private static ChangeItem PatchMeshNode( ISynchronizationStream stream, MeshNode current, JsonElement updated, JsonPatch? patch, string changedBy) { - var updatedNode = updated.Deserialize(stream.Hub.JsonSerializerOptions); - return new(updatedNode!, changedBy, stream.StreamId, ChangeType.Patch, stream.Hub.Version, + // One hub read, through RequireHub() β€” see the note on StandardReducers' patch functions + // (#3321 step 3): a reducer must return a change, so its only refusal channel is the + // TRANSIENT HubDisposingException, which ToChangeItem maps to its modelled `null`. + var hub = stream.RequireHub(); + var updatedNode = updated.Deserialize(hub.JsonSerializerOptions); + return new(updatedNode!, changedBy, stream.StreamId, ChangeType.Patch, hub.Version, [new EntityUpdate(nameof(MeshNode), updatedNode?.Id, updatedNode) { OldValue = current }]); } diff --git a/src/MeshWeaver.Layout/Composition/LayoutAreaHost.cs b/src/MeshWeaver.Layout/Composition/LayoutAreaHost.cs index 61681c91ee..09eef7aa5c 100644 --- a/src/MeshWeaver.Layout/Composition/LayoutAreaHost.cs +++ b/src/MeshWeaver.Layout/Composition/LayoutAreaHost.cs @@ -548,7 +548,15 @@ private void PushRenderResult(EntityStoreAndUpdates result) // Emit as a Full: a complete snapshot the client's control streams // re-evaluate wholesale, delivering nested sub-areas reliably. - return new ChangeItem(resultStore, Stream.StreamId, Stream.Hub.Version); + // 🚨 #3321 step 3: the hub is resolved ONCE, through the PRESENCE accessor, because + // this transform runs on the sync hub's turn β€” which a Dispose() on another thread can + // overtake β€” and a released stream has no Version to stamp. `null` is Update's own + // documented no-op. HubIfHeld, not TryGetHub: this mirrors the `TryGetActiveHub` guard + // Update already applied before posting the transform, so it refuses exactly when the + // write would be refused and never tightens a render that would still have landed. + if (Stream.HubIfHeld() is not { } renderHub) + return null; + return new ChangeItem(resultStore, Stream.StreamId, renderHub.Version); // resolvedArea, not Reference.Area β€” the latter is null for a default-area // subscription, so this line said "(null)" for exactly the renders #1182 was // about. The render-FAILURE paths were fixed there; these two remaining @@ -1428,7 +1436,10 @@ private void FailRendering(Exception ex, string? area) // Stop the "Building layout…" spinner now the area has resolved (to an error). .Update(LayoutAreaReference.Data, coll => coll.SetItem(ProgressDataId, new { message = "", progress = 100 })); - return new ChangeItem(store, Stream.StreamId, Stream.Hub.Version); + // Hub resolved through the presence accessor β€” see PushRenderResult (#3321 step 3). + if (Stream.HubIfHeld() is not { } errorHub) + return null; + return new ChangeItem(store, Stream.StreamId, errorHub.Version); }, updateEx => logger.LogWarning(updateEx, "Cannot surface render error for {Area}", area)); } @@ -1477,11 +1488,14 @@ public void UpdateProgress(string area, ProgressControl progress) /// Optional 0–100 completion hint. public void UpdateProgress(string message, double? percent = null) => Stream.Update(current => - new ChangeItem( - (current ?? new EntityStore()).Update(LayoutAreaReference.Data, - coll => coll.SetItem(ProgressDataId, new { message, progress = percent ?? 0 })), - Stream.StreamId, - Stream.Hub.Version), + // Hub resolved through the presence accessor β€” see PushRenderResult (#3321 step 3). + Stream.HubIfHeld() is not { } progressHub + ? null + : new ChangeItem( + (current ?? new EntityStore()).Update(LayoutAreaReference.Data, + coll => coll.SetItem(ProgressDataId, new { message, progress = percent ?? 0 })), + Stream.StreamId, + progressHub.Version), // resolvedArea β€” see "Cannot apply render for" above. ex => logger.LogWarning(ex, "Cannot update loading progress for {Area}", resolvedArea)); diff --git a/test/MeshWeaver.Data.Test/StreamLivenessIsConsumerReachableTest.cs b/test/MeshWeaver.Data.Test/StreamLivenessIsConsumerReachableTest.cs index b479f1a932..ef536b2d6c 100644 --- a/test/MeshWeaver.Data.Test/StreamLivenessIsConsumerReachableTest.cs +++ b/test/MeshWeaver.Data.Test/StreamLivenessIsConsumerReachableTest.cs @@ -97,11 +97,16 @@ public async Task ADisposedStream_IsNotUsable_EvenThoughHubStillAnswers() stream.Dispose(); - // 🚨 The assertion that names the defect. Hub is NOT null here β€” the contract's promise is - // kept to the letter β€” so nothing about `stream.Hub.Anything` looks wrong at a call site. - stream.Hub.Should().NotBeNull( - "the non-nullable contract still holds after disposal, which is precisely why a " - + "consumer needed a separate way to ask"); + // 🚨 This assertion INVERTED at step 3 (#3321), and the inversion is the whole point of the + // three-step order. Through steps 1 and 2 it read `Hub.Should().NotBeNull()` β€” the + // non-nullable contract was kept to the letter, so nothing about `stream.Hub.Anything` + // looked wrong at a call site, which is precisely why a consumer needed a separate way to + // ask. Step 3 is what made "absent" a state the field can be in, and it was only safe to + // ship BECAUSE steps 1 and 2 had landed: the accessor below existed, and every call site + // that could race a teardown had been migrated onto it. + stream.Hub.Should().BeNull( + "step 3 releases the hub on disposal β€” that release is what reclaims the leaked " + + "stream β†’ dead hub β†’ resolved state graph"); stream.IsUsable().Should().BeFalse("a disposed stream is a corpse"); stream.TryGetHub().Should().BeNull("the accessor is the way a consumer finds that out"); } diff --git a/test/MeshWeaver.Data.Test/StreamReleasesItsHubTest.cs b/test/MeshWeaver.Data.Test/StreamReleasesItsHubTest.cs new file mode 100644 index 0000000000..5edc8060ba --- /dev/null +++ b/test/MeshWeaver.Data.Test/StreamReleasesItsHubTest.cs @@ -0,0 +1,314 @@ +ο»Ώusing System; +using System.Reactive; +using System.Reactive.Linq; +using System.Runtime.CompilerServices; +using System.Text.Json; +using System.Threading.Tasks; +using MeshWeaver.Data.Serialization; +using MeshWeaver.Fixture; +using MeshWeaver.Json; +using MeshWeaver.Messaging; +using Xunit; + +namespace MeshWeaver.Data.Test; + +/// +/// 🚨 THE STREAM LETS THE HUB GO β€” Systemorph/MeshWeaver#3321, step 3 of 3, the step that actually +/// reclaims the leak. +/// +/// What was measured. Five heap dumps of a production replica (26 h uptime, 6.9 GB +/// working set, ~210 MB/h growth, 267 full collections freeing nothing) found 1 496 +/// MessageHubs at RunLevel=Dead still reachable β€” and the referrer walk over all +/// 51 562 911 objects named exactly ONE holder: SynchronizationStream<T>.Hub, 1 485 +/// <MeshNode> plus 11 <JsonElement>, one stream per corpse, exactly. No GC +/// root pointed at a dead hub directly. HostedHubsCollection had done its job (1 495 of 1 496 +/// entries removed) and the hubs' Autofac scopes had been closed β€” but closing a scope does not null +/// the fields of an object that outlives it, so each corpse kept its own ILifetimeScope and +/// TypeRegistry alive at ~390 KB apiece. +/// +/// 🚨 The split that decides what "fixing it" means. Of those 1 496, only 11 sat +/// under a stream that had itself been disposed β€” SynchronizationStream total 8 461, +/// disposed 11. So Dispose() clearing the field, which is how this step is usually +/// described, would have reclaimed 11 of 1 496. The other 1 485 are hosted sub-hubs killed by their +/// PARENT's teardown while the stream that created them β€” owned by a workspace somewhere else β€” was +/// never told. Both halves are tested here, and the second is the one that carries the bytes. +/// +/// Why this could not simply be done. The predecessor of the current constructor nulled +/// Hub and documented that "every code path that touches Hub goes through +/// TryGetActiveHub". That method was private to one file while ~96 sites dereferenced a +/// property declares NON-NULLABLE, so no consumer honoured it +/// and none could. It cost a production NRE inside LayoutAreaHost's constructor during a +/// recycle window, which reached the subscriber as a TERMINAL DeliveryFailure: a page told +/// "this failed forever" instead of "ask again". Steps 1 (#3380, make liveness askable) and 2 +/// (#3386, migrate the dereference sites) exist so that this step is no longer that change. See +/// Doc/Architecture/StreamLivenessAndTheHubReference. +/// +public class StreamReleasesItsHubTest(ITestOutputHelper output) : HubTestBase(output) +{ + private record Empty; + + /// + /// A data source, so + /// runs against a stream carrying the REAL PatchEntityStore reducer rather than a bare + /// stream with no PatchFunction β€” which would answer null for a reason that has + /// nothing to do with #3321, and so could not fail. + /// + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) + => base.ConfigureHost(configuration) + .AddData(data => data.AddSource(ds => + ds.WithType(type => type.WithKey(instance => instance.Id)))); + + private SynchronizationStream CreateStream(IMessageHub host) + => new( + new StreamIdentity(host.Address, null), + host, + new EntityReference("X", "Y"), + new ReduceManager(host), + null); + + /// + /// Half one: a stream that is disposed normally lets its hub go. + /// + /// 🚨 The assertion is a , not Hub is null, and that is + /// the point. "The field reads null" is a statement about one field; "the object is + /// unreachable" is the statement this issue is actually about, and it is the only one that + /// would have gone red on the defect for the RIGHT reason. The hub reference is created and + /// dropped inside a non-inlined helper so no JIT-visible local can keep it alive on the + /// test's own frame. + /// + [HubFact] + public async Task ADisposedStream_ReleasesItsHub_SoTheHubBecomesCollectable() + { + var host = GetHost(); + var (stream, hubRef, disposalCompleted) = CreateDisposeAndWeaklyReference(host); + + stream.Hub.Should().BeNull( + "the stream must drop the reference on disposal β€” until step 3 it kept the corpse and " + + "everything the corpse had resolved reachable for the process's whole life"); + stream.TryGetHub().Should().BeNull("and the guarded accessor agrees"); + + // 🚨 WAIT FOR THE HUB'S OWN TEARDOWN TO FINISH BEFORE COLLECTING, and the reason is a + // measurement, not caution. `Hub.Dispose()` only STARTS the teardown β€” it posts a + // ShutdownRequest and returns β€” so the instant `Dispose()` returns, the hub is still rooted + // by its own in-flight shutdown (its action block, its scheduler, its registry entry). The + // first version of this test collected immediately and asserted unreachability. It PASSED + // in a full-suite run (where the surrounding tests happened to give the teardown time) and + // FAILED run in isolation β€” a lucky green over an assertion the test never actually + // observed, which is the one failure mode a reachability test must not have. Waiting on + // DisposalCompleted makes the claim deterministic: after it, the ONLY thing that could + // still hold the hub is a reference somebody kept β€” which is exactly what this asserts. + await disposalCompleted + .Catch(_ => Observable.Return(Unit.Default)) + .FirstOrDefaultAsync() + .Timeout(TestTimeouts.Convergence) + .Await(TestContext.Current.CancellationToken); + + Collect(); + + hubRef.IsAlive.Should().BeFalse( + "the hub must be UNREACHABLE, not merely unreferenced by one field β€” 1 496 dead hubs " + + "were held alive in production by exactly this reference and nothing else"); + } + + /// + /// 🚨 Half two, and the half that carries 1 485 of the 1 496: the hub dies UNDER a stream that + /// nobody disposes. This is the production shape β€” a sync/ hub is a HOSTED hub, so a + /// Blazor circuit ending, a DisposeRequest or a recycle disposes it through the parent's + /// teardown, while the stream that created it is owned by a workspace elsewhere and is never + /// notified. A step 3 that only cleared the field in Dispose() would leave this case + /// exactly as it was, which is why it is tested separately rather than folded into the one + /// above. + /// + [HubFact] + public async Task AHubDyingUnderAnUndisposedStream_IsStillReleased() + { + var host = GetHost(); + var stream = CreateStream(host); + var hub = stream.Hub; + hub.Should().NotBeNull("precondition: the stream owns a live sync hub"); + + // Kill the hub the way a parent teardown does β€” WITHOUT disposing the stream. This is the + // very call Workspace.EvictClientSubscriptions makes. + hub.Dispose(); + await hub.DisposalCompleted + .Catch(_ => Observable.Return(Unit.Default)) + .FirstOrDefaultAsync() + .Timeout(TestTimeouts.Convergence) + .Await(TestContext.Current.CancellationToken); + + stream.Hub.Should().BeNull( + "the stream learns of its hub's death through the hook registered on the hub itself β€” " + + "1 485 of the 1 496 corpses in the production dump were exactly this case, held by a " + + "stream that had never been disposed and never would be"); + stream.IsUsable().Should().BeFalse( + "and a released stream reports itself unusable through the SAME predicate every cache " + + "already consults β€” no new state, no second liveness answer"); + } + + /// + /// The control, and it is not decoration: every assertion above is satisfied by a stream that + /// releases its hub immediately and never works. This pins that a LIVE stream still answers + /// with its hub, through both accessors, and still writes. + /// + [HubFact] + public void ALiveStream_StillAnswersWithItsHub() + { + var host = GetHost(); + var stream = CreateStream(host); + + stream.Hub.Should().NotBeNull(); + stream.TryGetHub().Should().BeSameAs(stream.Hub, + "the guarded accessor resolves the very hub the non-nullable property does"); + stream.RequireHub().Should().BeSameAs(stream.Hub, + "and so does the require-accessor the patch reducers use"); + + Action act = () => stream.Update(_ => new Empty(), _ => { }); + act.Should().NotThrow("a live stream still takes writes"); + } + + /// + /// refuses with the TRANSIENT + /// , never a . + /// + /// That difference is the whole reason step 3 is safe where the predecessor's attempt was + /// not. is an , so it + /// classifies as ErrorType.ShuttingDown β€” "the address may reactivate, ask again" β€” and + /// the Blazor circuit's existing catch around BindStream() already handles it. The raw + /// NRE the predecessor produced classified as a TERMINAL DeliveryFailure instead. + /// + [HubFact] + public void RequireHub_OnAReleasedStream_RefusesTransiently() + { + var host = GetHost(); + var stream = CreateStream(host); + stream.Dispose(); + + Action act = () => _ = stream.RequireHub(); + + var ex = act.Should().Throw( + "a surface that must return a value has only a throw for a refusal, and this one is the " + + "transient shape the framework already classifies as retryable"); + ex.Which.HubAddress.Should().Be(host.Address, + "the refusal names the host so a caller can retry the right address"); + (ex.Which is ObjectDisposedException).Should().BeTrue( + "staying in the ObjectDisposedException family is what keeps the existing teardown-aware " + + "handling (Blazor's BindStream catch, SynchronizationStream.OnError) working"); + } + + /// + /// 🚨 The reducers' single gateway absorbs that refusal into the null its own signature + /// already models. Every patch reducer in the codebase β€” StandardReducers' three, + /// MeshDataSource.PatchMeshNode β€” is invoked from exactly one place, + /// ToChangeItem, and null is what it already returns for a stream with no + /// PatchFunction registered. So a reducer racing a disposal degrades to "no patch + /// derived", which both callers already handle, instead of to a fault. + /// + [HubFact] + public async Task ToChangeItem_OnAReleasedStream_AnswersWithItsModelledAbsence() + { + var host = GetHost(); + var workspace = host.GetWorkspace(); + var collection = workspace.DataContext.GetTypeSource(typeof(MyData))!.CollectionName; + var stream = workspace.GetStream(new CollectionsReference(collection)); + ((object?)stream).Should().NotBeNull(); + + // Let the stream reach a state a patch could apply to, so the negative below is about the + // RELEASE and not about a stream that never started. + var current = await stream!.Timeout(TestTimeouts.Convergence) + .FirstAsync() + .Await(TestContext.Current.CancellationToken); + + // An EMPTY patch on purpose: PatchEntityStore's operation loop then does nothing and the + // method runs straight to its Version stamp β€” the read that has to answer for itself. A + // populated patch would exercise the deserialization arms too and blur which read the test + // is about. + var json = JsonSerializer.SerializeToElement(current.Value, stream!.Hub.JsonSerializerOptions); + var emptyPatch = new JsonPatch(); + + stream.ToChangeItem(current.Value!, json, emptyPatch, "test") + .Should().NotBeNull("precondition: a LIVE stream still derives a change"); + + stream.Dispose(); + + stream.ToChangeItem(current.Value!, json, emptyPatch, "test") + .Should().BeNull( + "a released stream derives nothing β€” and says so through the absent value this " + + "method already returns when no PatchFunction is registered, never by NRE'ing " + + "inside a reducer that has no way to answer 'no'"); + } + + /// + /// 🚨 THE TWO ACCESSORS ANSWER DIFFERENT QUESTIONS, and this pins the difference β€” because + /// collapsing them looks like a tidy-up and is a behaviour change. + /// + /// A FAULTED stream still HOLDS its hub. TryGetHub() refuses it, correctly: it + /// asks "should I use this stream?", and counts a terminal store + /// error exactly as much as a disposal (#2387). HubIfHeld() hands the hub back, also + /// correctly: it asks "is the reference still there?", and it is. + /// + /// Why the distinction is not academic. Step 3's first draft guarded + /// DataSource.Initialized β€” Task.WhenAll(streams.Select(s => s.Hub.Started)) β€” + /// with TryGetHub(). A stream whose initial load faulted was therefore dropped from the + /// WhenAll, so Initialized completed SUCCESSFULLY and a hub whose data source had thrown + /// went on answering requests as though nothing had happened. + /// DataContextFaultedInitBeforeStreamHubBoundTest caught it; this test is what stops the + /// same substitution being made again without one. + /// + [HubFact] + public void AFaultedStream_StillHoldsItsHub_EvenThoughItIsNotUsable() + { + var host = GetHost(); + var stream = CreateStream(host); + + stream.OnError(new InvalidOperationException("boom")); + + stream.IsUsable().Should().BeFalse( + "a terminal store error is as permanent as a disposal β€” a ReplaySubject replays it to " + + "every later subscriber forever (#2387)"); + stream.TryGetHub().Should().BeNull( + "so a CONSUMER asking 'should I use this stream?' must be told no"); + stream.HubIfHeld().Should().NotBeNull( + "but the reference is still THERE, and an owner-side guard that asks the presence " + + "question must not be answered as though it had asked the liveness one β€” that " + + "substitution turned a faulted data-source init into a silently successful one"); + stream.RequireHub().Should().BeSameAs(stream.Hub, + "and the throwing form asks the same presence question, so it does not throw either"); + } + + /// + /// Forces the collection the reachability assertion reads. Two passes with a finalizer drain + /// between them: the first may only queue the object for finalization, and a hub holds + /// finalizable state. + /// + private static void Collect() + { + for (var i = 0; i < 3; i++) + { + GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, blocking: true); + GC.WaitForPendingFinalizers(); + } + } + + /// + /// Builds a stream, weakly references its hub, captures the hub's completion signal, and + /// disposes the stream β€” all inside a frame that is GONE by the time the caller collects, so no + /// live local on the test's own stack can root the hub and make a red test look green. + /// + /// The returned DisposalCompleted is safe to hold: it is the hub's + /// AsObservable() wrapper around a subject FIELD, and a subject does not reference the + /// object that owns it. Returning the hub itself, or any closure over it, would defeat the + /// whole test. + /// + [MethodImpl(MethodImplOptions.NoInlining)] + private (SynchronizationStream Stream, WeakReference HubRef, IObservable DisposalCompleted) + CreateDisposeAndWeaklyReference(IMessageHub host) + { + var stream = CreateStream(host); + var hub = stream.Hub; + var hubRef = new WeakReference(hub); + var disposalCompleted = hub.DisposalCompleted; + hubRef.IsAlive.Should().BeTrue("precondition: the stream owns a live sync hub"); + stream.Dispose(); + return (stream, hubRef, disposalCompleted); + } +} diff --git a/test/MeshWeaver.Layout.Test/TornDownStreamCallSitesTest.cs b/test/MeshWeaver.Layout.Test/TornDownStreamCallSitesTest.cs index 03ffccf967..8f59e7802f 100644 --- a/test/MeshWeaver.Layout.Test/TornDownStreamCallSitesTest.cs +++ b/test/MeshWeaver.Layout.Test/TornDownStreamCallSitesTest.cs @@ -71,9 +71,17 @@ await stream.GetControlStream(TestArea) stream.IsUsable().Should().BeTrue("precondition: the stream served the area before we kill it"); stream.Dispose(); stream.TryGetHub().Should().BeNull("precondition: the stream is a corpse now"); - stream.Hub.Should().NotBeNull( - "and the non-nullable contract still holds β€” which is exactly why every site below " - + "looked safe while it was not"); + // 🚨 This assertion INVERTED at step 3 (#3321). Through step 2 it read + // `Hub.Should().NotBeNull()` β€” "the non-nullable contract still holds, which is exactly why + // every site below looked safe while it was not". Step 3 is the change that makes it stop + // holding: a disposed stream now RELEASES the hub, which is what reclaims the leaked + // `stream β†’ dead hub β†’ resolved state` graph. Keeping the assertion (flipped) rather than + // deleting it is deliberate β€” it is the line that pins WHICH of the two contracts is in + // force, and a silent revert of step 3 turns it red here as well as in + // StreamReleasesItsHubTest. + stream.Hub.Should().BeNull( + "step 3 releases the reference on disposal β€” the declaration stays non-nullable for the " + + "packaged interface, and what makes that safe is that every site below can now ask"); return stream; }