Reclaim change-feed-evicted mirrors instead of parking them forever (#1324)

src/MeshWeaver.Data/Workspace.cs

@@ -130,11 +130,13 @@ private void EvictForPath(string path)
         if (string.IsNullOrEmpty(path) || _remoteStreamCache.IsEmpty)
             return;
 
-        // Do NOT dispose the evicted stream — existing subscribers (e.g. live Blazor
-        // components) are still attached to it and need to keep receiving updates
-        // until they drop on their own. The eviction only prevents NEW callers from
-        // re-using the now-stale stream; the next GetRemoteStream creates a fresh one
-        // against the (re-)activated owner.
+        // Do NOT unconditionally dispose the evicted stream — an undeclared reader (e.g. a
+        // MeshDataSource reduce callback that handed the stream on) may still be attached and
+        // needs to keep receiving updates until it drops on its own. The eviction only prevents
+        // NEW callers from re-using the now-stale stream; the next GetRemoteStream creates a
+        // fresh one against the (re-)activated owner. A stream whose holders DECLARED
+        // themselves (see <see cref="LeaseRemoteStream"/>) and have all left is reclaimed
+        // immediately — it is out of the cache, so nothing can adopt it any more.
         foreach (var key in _remoteStreamCache.Keys)
         {
             // Owner-address match only — every identity's stream for that owner is evicted.
@@ -146,14 +148,124 @@ private void EvictForPath(string path)
                 // disposed → TimerQueue-pinned forever). Only a materialised stream
                 // has a hub to dispose.
                 if (removed.IsValueCreated)
+                {
                     _evictedRemoteStreams[removed.Value] = 0;
+                    ReclaimIfUnheld(removed.Value);
+                }
                 _logger.LogDebug(
                     "Evicted remote stream cache for {Address} after change event.",
                     key.Owner);
             }
         }
     }
 
+    // 🚨 THE LIFETIME OF AN EVICTED STREAM — declared holders, NOT Rx subscriber counts.
+    //
+    // A change-feed eviction takes a stream OUT of _remoteStreamCache but cannot dispose it at
+    // the eviction site: it does not know whether anyone is still reading it. Parking it and
+    // hoping is what leaked — every write to a subscribed path minted a fresh client `sync/` hub
+    // (and, via its SubscribeRequest, a matching `sync/` hub on the owner) while every
+    // predecessor sat in _evictedRemoteStreams until the process died (Systemorph/MeshWeaver#1324:
+    // the parked set grew 1 → 23 monotonically over three NodeType recompiles and was never
+    // drained — its only reaper is the shared mesh-node cache's idle sweep, which needs zero
+    // subscribers AND ten minutes untouched, a condition a continuously-written path never meets).
+    //
+    // 🚨 Counting Rx subscribers CANNOT answer "is anyone still reading this": the reduce chain
+    // CreateExternalClient builds subscribes to the stream ITSELF (the outbound
+    // change-notification pipeline and the version-gate observer), so an evicted stream measures
+    // 2–3 subscribers and never reaches zero. That approach was implemented, measured and
+    // reverted — a mechanism that never triggers is worse than none.
+    //
+    // So holders DECLARE themselves. Everything that keeps a remote stream past the call that
+    // resolved it takes a LEASE (AcquireRemoteStreamUnchecked) and releases it when it is done — the
+    // shared mesh-node cache's hydration for as long as its entry lives, and each cross-hub write
+    // for the duration of its Observable.Create subscription. When the last declared holder
+    // leaves a stream that is already evicted, nothing can adopt it (it is out of the cache), so
+    // it is disposed at once: UnsubscribeRequest goes to the owner, both `sync/` hubs die.
+    //
+    // A stream NOBODY leased is never in this registry and keeps the old conservative parking —
+    // undeclared holders (the MeshDataSource / SyncedQueryDataSource reduce callbacks) are
+    // unaffected. Opting a call site in is one line and is what makes its streams reclaimable.
+    private readonly ConcurrentDictionary<ISynchronizationStream, int> _remoteStreamLeases =
+        new(StreamReferenceComparer.Instance);
+
+    /// <summary>
+    /// Resolves the remote stream for (<paramref name="owner"/>, <paramref name="reference"/>)
+    /// AND declares the caller as a holder of it. Dispose the returned lease when done — that is
+    /// what lets an evicted stream be reclaimed instead of parked forever (see the
+    /// <see cref="_remoteStreamLeases"/> note). The liveness re-check closes the window where the
+    /// stream was reclaimed between resolution and the lease: a dead instance is released and the
+    /// resolve retried, which builds a fresh one (the same retry contract as
+    /// <see cref="GetExternalClientSynchronizationStream{TReduced,TReference}"/>).
+    /// </summary>
+    internal (ISynchronizationStream<TReduced> Stream, IDisposable Lease)
+        AcquireRemoteStreamUnchecked<TReduced, TReference>(Address owner, TReference reference)
+        where TReference : WorkspaceReference
+    {
+        while (true)
+        {
+            var stream = GetRemoteStreamUnchecked<TReduced, TReference>(owner, reference);
+            var lease = LeaseRemoteStream(stream);
+            if (stream.Hub?.RunLevel <= MessageHubRunLevel.Started
+                && stream.Hub is not MessageHub { IsDisposing: true })
+                return (stream, lease);
+            lease.Dispose();
+        }
+    }
+
+    /// <summary>Declares a holder of <paramref name="stream"/>; disposing the returned
+    /// handle releases it (idempotent).</summary>
+    private IDisposable LeaseRemoteStream(ISynchronizationStream stream)
+    {
+        _remoteStreamLeases.AddOrUpdate(stream, 1, (_, count) => count + 1);
+        // Disposable.Create runs its action AT MOST ONCE (Interlocked-swapped internally), so a
+        // double-dispose of the lease handle cannot under-count the holders.
+        return System.Reactive.Disposables.Disposable.Create(() =>
+        {
+            while (true)
+            {
+                // 🚨 Read-then-TryUpdate, never AddOrUpdate: DetachRemoteStreams REMOVES the
+                // bookkeeping when it hands ownership out, and re-adding a zero entry here would
+                // pin a stream this workspace no longer owns.
+                if (!_remoteStreamLeases.TryGetValue(stream, out var current))
+                    return;
+                var remaining = current > 0 ? current - 1 : 0;
+                if (!_remoteStreamLeases.TryUpdate(stream, remaining, current))
+                    continue;
+                if (remaining == 0)
+                    ReclaimIfUnheld(stream);
+                return;
+            }
+        });
+    }
+
+    /// <summary>
+    /// Disposes <paramref name="stream"/> iff it is BOTH evicted (parked — no caller can adopt
+    /// it) AND has no remaining declared holder. Called from the two edges that can make that
+    /// true: the eviction itself, and the release of the last lease.
+    /// </summary>
+    private void ReclaimIfUnheld(ISynchronizationStream stream)
+    {
+        if (!_remoteStreamLeases.TryGetValue(stream, out var leases) || leases != 0)
+            return;
+        if (!_evictedRemoteStreams.TryRemove(stream, out _))
+            return;
+        _remoteStreamLeases.TryRemove(stream, out _);
+        try
+        {
+            stream.Dispose();
+            _logger.LogDebug(
+                "Workspace {WorkspaceId} disposed superseded remote stream for {Owner} — "
+                + "no declared holder remains.", Id, stream.Owner);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogDebug(ex,
+                "Workspace {WorkspaceId} error disposing superseded remote stream for {Owner}",
+                Id, stream.Owner);
+        }
+    }
+
     /// <summary>
     /// Opens the initialization gate after all handlers are registered.
     /// Called via SyncBuildupActions to ensure proper ordering.
@@ -399,6 +511,12 @@ internal IReadOnlyList<ISynchronizationStream> DetachRemoteStreams(
                 && _evictedRemoteStreams.TryRemove(parked, out _))
                 detached.Add(parked);
         }
+        // Ownership moves to the caller (it disposes, or hands them back via ParkRemoteStreams),
+        // so drop the lease bookkeeping: keeping a zero-count entry would pin a stream this
+        // workspace no longer owns. A re-parked stream simply re-enters the conservative
+        // "undeclared holders" bucket until a fresh lease is taken.
+        foreach (var stream in detached)
+            _remoteStreamLeases.TryRemove(stream, out _);
         return detached;
     }
 
@@ -440,8 +558,19 @@ private ISynchronizationStream<TReduced> GetExternalClientSynchronizationStream<
                     () => (ISynchronizationStream)this.CreateExternalClient<TReduced, TReference>(address, reference),
                     LazyThreadSafetyMode.ExecutionAndPublication));
 
+            // The counterpart of the eviction line below, and the one #1324 needed: a client
+            // `sync/` hub (plus its owner-side twin) is born here, so "who keeps minting mirrors
+            // for this path, under which identity" is answerable from a Debug-level run instead of
+            // a heap dump. Logged only on a genuine MISS — a cache hit is free and silent.
+            var freshlyCreated = !lazy.IsValueCreated;
+
             var stream = lazy.Value;
 
+            if (freshlyCreated)
+                _logger.LogDebug(
+                    "Workspace {WorkspaceId} opened remote stream {StreamId} for {Owner} as {Identity}.",
+                    Id, stream.StreamId, key.Item1, key.Item3);
+
             // Check if cached stream is still alive. Hub.RunLevel alone is not
             // sufficient because hub shutdown is async: right after stream.Dispose()
             // is called, RunLevel is still Running even though disposal was triggered.
@@ -674,6 +803,8 @@ public void Dispose()
                 _logger.LogDebug(ex, "Workspace {WorkspaceId} error disposing evicted remote stream", Id);
             }
         }
+        // Nothing left to reclaim — drop the holder bookkeeping so it stops referencing streams.
+        _remoteStreamLeases.Clear();
 
         // Local reduced streams are OWNED by their parent (CreateReducedStream registers each on
         // the data-source stream that produced it), so DataContext.Dispose below tears them down.

src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md

@@ -151,6 +151,35 @@ NodeType activation, MCP get/search, synced-query grain warming — leaked a
 permanently-connected upstream stream (~1,650 live streams / 37 heartbeats-per-second
 measured on a long-lived portal).
 
+### The idle sweep is not the only reaper — an evicted upstream dies with its last holder
+
+The idle sweep answers *"this path went quiet"*. It cannot answer *"this write is finished
+with its mirror"*, and that is the case a busy path is always in: `Workspace.EvictForPath`
+retires a path's upstream on **every** change event — including the echo of the writer's own
+write — so the next writer diffs against the owner's authoritative state rather than a stale
+snapshot. That eviction is load-bearing; parking the retired stream until the idle sweep
+happened to notice was not, and a continuously-written path never meets the sweep's
+"zero subscribers **and** ten minutes untouched" condition
+([#1324](https://github.com/Systemorph/MeshWeaver/issues/1324)).
+
+Reference counting the stream's Rx subscribers cannot decide it either: the reduce chain
+`JsonSynchronizationStream.CreateExternalClient` builds subscribes to the stream **itself**, so
+a retired mirror sits at 2–3 subscribers forever and a "dispose at zero" rule never fires.
+
+So holders **declare** themselves. Anything keeping a remote stream past the call that resolved
+it takes a lease from `Workspace.AcquireRemoteStreamUnchecked` and disposes it when done:
+
+| holder | lease lifetime |
+|---|---|
+| this cache's per-entry hydration (`CreateEntry` → `handle.Subscribe`) | the entry's whole life — it *is* the path's one live mirror |
+| a cross-hub write (`MeshNodeStreamHandle.UpdateRemote` / `WriteViaSyncStream`) | the write's `Observable.Create` subscription |
+| a reduce callback that hands the stream on (`MeshDataSource`, `SyncedQueryDataSourceExtensions`) | **none** — undeclared, so it keeps the conservative parking |
+
+An **evicted** stream with no remaining declared holder is disposed at once — `UnsubscribeRequest`
+reaches the owner, and both the client and owner `sync/` hubs die. A stream nobody leased is never
+touched by this and is still reclaimed by the idle release above or at workspace disposal. Steady
+state on a hot path is therefore ONE live mirror, not one per write.
+
 ## The storm breaker — absent nodes can't melt the silo
 
 A read whose owner answers *NotFound* is cached as a **negative entry** with exponential backoff (2 s doubling up to 5 min). While the window is open, re-subscribing to that path replays the cached failure instead of re-opening an upstream subscription — so a loop that keeps re-reading an absent node cannot hammer the routing layer. The entry simply **expires** (the next natural read re-probes once; a successful read clears it immediately) — there is no timer that re-subscribes on its own.

src/MeshWeaver.Documentation/Data/DataMesh/SyncedQueryDataSource.md

@@ -195,6 +195,20 @@ This is exactly what the `MeshNodeReference(path)` reducer registered by `AddSyn
 - When the cached stream's hub leaves `MessageHubRunLevel.Started` (disposed or failed), the next `GetRemoteStream` call replaces it with a fresh instance.
 - When an explicit `IMeshChangeFeed` event reports a path change (ownership churn — see `Workspace.EvictForPath`), open subscribers stay attached to their existing stream while new callers bind to a fresh one tied to the re-activated owner.
 
+**An evicted stream is closed as soon as its last declared holder lets go.** Eviction removes the
+stream from the cache but cannot dispose it at the eviction site — it does not know whether anyone
+is still reading. Rx subscriber counts cannot answer that either: the reduce chain
+`CreateExternalClient` builds subscribes to the stream itself, so an evicted stream sits at 2–3
+subscribers forever. So holders **declare** themselves: anything that keeps a remote stream past the
+call that resolved it takes one through `Workspace.AcquireRemoteStreamUnchecked` and disposes the
+returned lease when it is done — the shared mesh-node cache's hydration for its entry's whole life,
+each cross-hub write for the duration of its `Observable.Create` subscription. When the last lease on
+an already-evicted stream goes, it is disposed at once: `UnsubscribeRequest` reaches the owner and
+both `sync/` hubs die. A stream **nobody** leased keeps the conservative parking (it may still have
+undeclared readers) and is reclaimed by the mesh-node cache's idle release or at workspace disposal.
+Without the leases every write to a subscribed path left a client `sync/` hub and its owner-side twin
+behind for the process lifetime — [#1324](https://github.com/Systemorph/MeshWeaver/issues/1324).
+
 The full integrity story is a plain dictionary, single-threaded access, and shared-by-default semantics — made safe entirely by the actor model.
 
 The `RemoteStreamCacheTest` (`test/MeshWeaver.Query.Test`) pins this contract: two `GetRemoteStream(...)` calls for the same key are reference-equal, and a disposed stream is evicted before the next caller receives it.

src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-writing-to-a-page-no-longer-leaves-a-connection-behind.md

@@ -0,0 +1,38 @@
+---
+Name: Writing to a page no longer leaves a connection behind
+Category: Fix
+Description: Every save to a busy item used to abandon the internal connection it wrote through and open a fresh one, so the portal accumulated connections for as long as it ran. Abandoned connections are now closed the moment nobody is using them.
+Icon: Sparkle
+Order: -20260813
+---
+
+# Writing to a page no longer leaves a connection behind
+
+When one part of the portal reads or writes something owned by another part, it opens a private
+live connection to it — a mirror that stays in step while it is needed. Mirrors are shared, so a
+page being read by ten things costs one mirror, not ten.
+
+There is one moment when a mirror has to be replaced rather than reused: the item changed, and the
+next writer must diff against what the owner actually holds, not against a snapshot from before.
+The portal handled that by retiring the mirror so the next caller would build a fresh one. Retiring
+it was correct. What was missing was closing it.
+
+Nothing could close it safely, because nothing knew whether anyone was still reading. Counting
+readers does not answer the question — a mirror's own internal plumbing subscribes to it, so the
+count never reaches zero even when every real reader has gone. So a retired mirror was simply
+parked, and parked mirrors were only ever cleaned up when the item went untouched for ten minutes
+— which an item being written to continuously never does.
+
+The result: every save to a busy item retired one mirror and opened another, on both ends of the
+connection, and neither of the retired pair ever went away. A single rebuild of in-portal code left
+about twenty of these behind; a morning of them left thousands.
+
+Holders now say so. Anything that keeps a mirror past the moment it asked for one — a live view
+being read, a save in flight — declares itself, and a retired mirror is closed the instant its last
+declared holder lets go. Measured over a rebuild loop: of twenty-four mirrors opened and
+twenty-three retired, eighteen are now closed immediately and the survivors are exactly the ones
+still being read. A rebuild costs roughly half the internal machinery it did this morning, and the
+saving applies to every ordinary save, not only to rebuilds.
+
+Two smaller sources of the same kind of growth remain, both with different causes, and the
+measurement that found this one is tightened again so neither can hide.