From 2ea66e08ab92fd13233e0754b32533533eea761f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Thu, 13 Aug 2026 16:24:11 +0200 Subject: [PATCH 1/2] fix: a finished activity was a KEEP-ALIVE, not something nobody had collected yet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the last residual of #1324 — the per-compile `_Activity/compile-` node hubs. The previous pass left the question open as "is this a permanent leak or a ~25-minute bounded retention", because both existing reclaimers are time-based and invisible to a seconds-long test. The framing was wrong in the consumer's favour. A cross-hub write opens a shared IMeshNodeStreamCache mirror on the path, and that mirror posts a HeartBeatEvent to the owning hub every 45 s FOR THE EXPRESS PURPOSE of keeping it alive — which re-arms every idle clock the platform has: the cache's own ten-minute window, an Orleans grain's collection age, KernelContainer.DisposeOnTimeout. So a finished compile was not waiting to be reclaimed; it was preventing its own reclamation, and holding its node hub and both sides' sync sub-hubs up with it. The cure is the missing EVENT, not a shorter timer. An ActivityLog that reaches a status where IsTerminal() holds will never be written again — by definition — so ActivityLogAppender.Append releases the path's warm entry as part of that same write. Nothing about the idle heuristic changes; what changed is that a caller who KNOWS the answer no longer waits for the heuristic to guess it. IMeshNodeStreamCache.ReleaseIfUnwatched(path) is the idle sweep's own detach → claim → tear-down protocol with TimeSpan.Zero in place of the window; both callers now share one implementation. The atomic zero-subscriber guard is unchanged, so a reader still watching the finished activity wins the race and keeps its mirror — the release simply declines. Released on onCompleted, not onNext: tearing the upstream down while the write is still in flight would strand its own terminal. Measured on NodeTypeRecompileAlcLeakTest, matched runs from an identical 38-hub baseline with four retained activities: hubs per compile activity 6.5 -> 5.0 of which mirror-side 3.0 -> 0 hubs per recompile (total) 8.7 -> 6.0 and the direct property — no terminal activity still holds a warm mirror — now holds; it fails before the change, with all four still warm after a 30 s wait. The ratchet tightens 9 -> 8, above the measured 6.0/6.7 and below the 8.7 the un-released mirror produced, so a regression of THIS fix reds the test. What is left is the activity's own node hub plus the four sync sub-hubs its data source builds at startup. Nothing subscribes to it any more, so on Orleans it is finally eligible for the idle deactivation the heartbeat used to keep pushing away; the monolith has no idle collection for node hubs at all, which is what the test still counts. That is a host-lifecycle question, and not one to "fix" by shortening a timer. Verified locally, -c Release -warnaserror: - MeshWeaver.Hosting.Monolith.Test (full) - MeshWeaver.Documentation.Test 20/20 Co-Authored-By: Claude Opus 5 (1M context) --- .../Data/Architecture/MeshNodeStreamCache.md | 26 +++ ...-a-finished-activity-lets-go-of-its-hub.md | 26 +++ src/MeshWeaver.Hosting/MeshNodeStreamCache.cs | 87 +++++++--- .../ActivityLogAppender.cs | 83 ++++++++-- .../Services/IMeshNodeStreamCache.cs | 25 +++ .../IndeterminateProbeReResolutionTest.cs | 1 + .../NodeTypeEnrichmentDoubleCallTest.cs | 1 + .../NodeTypeRecompileAlcLeakTest.cs | 151 +++++++++++++++--- 8 files changed, 337 insertions(+), 63 deletions(-) create mode 100644 src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-finished-activity-lets-go-of-its-hub.md diff --git a/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md b/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md index 8e92e5d0cc..39e30c29f7 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md @@ -202,6 +202,32 @@ 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). +### A warm mirror is a KEEP-ALIVE, so "final" must be an event, not a wait + +The ten-minute window is a *heuristic about the future*: a path touched just now is +probably about to be touched again. For a node with a **terminal lifecycle** that +heuristic is provably false, and waiting it out is not merely wasteful — it is +self-defeating. The mirror posts a `HeartBeatEvent` to its owner every 45 s +(`SyncStreamOptions.HeartbeatInterval`) *expressly to keep that hub alive*, and that +message re-arms every idle clock the platform has: the cache's own window, an Orleans +grain's collection age, `KernelContainer.DisposeOnTimeout`. So a finished activity does +not sit waiting to be reclaimed — it **prevents its own reclamation**, and it takes the +owning node hub and the `sync/` sub-hubs on both sides with it +([#1324](https://github.com/Systemorph/MeshWeaver/issues/1324)). + +`IMeshNodeStreamCache.ReleaseIfUnwatched(path)` is the event-driven counterpart: the same +detach → claim → tear-down protocol as the sweep, and the same **atomic zero-subscriber +guard**, but with the wait replaced by proof from the caller. Its one caller today is +`ActivityLogAppender.Append`, which releases the path in the same write that reports a +status where `ActivityStatus.IsTerminal()` holds — an activity in that state is never +written again, by definition. A reader still watching the finished activity wins the race +and keeps its mirror; the release simply declines and the sweep remains the backstop. + +🚨 It is **not** a shorter timer, and must not become one. Nothing about the idle window +changes; what changed is that a caller who *knows* the answer no longer has to wait for +the heuristic to guess it. Measured on `NodeTypeRecompileAlcLeakTest`: 6.5 → 5.0 retained +hubs per compile activity, with the `cache/`-side mirror hubs going 3 → 0. + ### 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 diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-finished-activity-lets-go-of-its-hub.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-finished-activity-lets-go-of-its-hub.md new file mode 100644 index 0000000000..905ea2b19d --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-finished-activity-lets-go-of-its-hub.md @@ -0,0 +1,26 @@ +--- +Name: A finished activity lets go of its hub +Category: Fix +Description: A compile, import or export that has ended no longer holds its own activity node open. The keep-alive it used to leave behind is what made a busy morning of recompiles grow the portal without bound. +Icon: ArrowSyncCheckmark +Order: -20260813 +--- + +# A finished activity lets go of its hub + +Every compile, import, export and mirror writes its progress into an activity node, and the write +opens a shared connection to that node so the next line can be appended cheaply. When the activity +ended, that connection stayed open. + +It looked harmless — the platform already releases connections that have gone quiet, and node hubs +already shut down when they go idle. But an open connection sends a heartbeat to the node it is +connected to, every 45 seconds, precisely so that node stays up. So a finished activity was not +waiting to be tidied away: it was **holding itself open**, and holding its node's hub open with it. +On a morning of merges — each one re-importing content and recompiling the types it touched — the +portal accumulated one of those per compile and never gave any of them back. + +Now the write that records the final status also lets go of the connection. Nothing else changes: +if you have the finished activity open in front of you, your view keeps its live connection and +nothing is torn down under you; and if the release does not happen for any reason, the existing +quiet-path cleanup still gets there. Measured on the recompile repro, hubs retained per compile +activity drop from 6.5 to 5.0, and the connection-side share of that goes to zero. diff --git a/src/MeshWeaver.Hosting/MeshNodeStreamCache.cs b/src/MeshWeaver.Hosting/MeshNodeStreamCache.cs index 07eba70bce..dcc076650d 100644 --- a/src/MeshWeaver.Hosting/MeshNodeStreamCache.cs +++ b/src/MeshWeaver.Hosting/MeshNodeStreamCache.cs @@ -967,32 +967,7 @@ private void ReleaseIdleReadStreams() try { foreach (var (path, lazy) in _streams) - { - if (!lazy.IsValueCreated) - continue; - var entry = lazy.Value; - if (!entry.IsIdleCandidate(readStreamIdleExpiration)) - continue; - - var detached = entry.Handle.DetachUpstreams(); - if (!entry.TryMarkIdleEvicted(readStreamIdleExpiration)) - { - // Lost the race — a consumer pinned the entry between the - // pre-check and the mark. Hand the upstreams back; the entry - // keeps serving them. (A later write may open a fresh upstream - // alongside — the same benign divergence a change-feed evict - // produces; both are reaped on the eventual release.) - entry.Handle.ReparkUpstreams(detached); - continue; - } - - _streams.TryRemove(new KeyValuePair>(path, lazy)); - var upstreamReleased = TearDownEntry(path, entry, detached); - logger.LogDebug( - "MeshNodeStreamCache: released idle read stream for {Path} (upstreams disposed: {Count})", - path, detached.Count); - readStreamEvictions.OnNext(new ReadStreamEviction(path, upstreamReleased, "idle")); - } + TryReleaseUnwatched(path, lazy, readStreamIdleExpiration, "idle"); } catch (Exception ex) { @@ -1002,6 +977,66 @@ private void ReleaseIdleReadStreams() } } + /// + /// The detach-then-mark release protocol, shared by the timed idle sweep and the event-driven + /// . The ONLY difference between the two callers is + /// — the sweep waits out the ten-minute heuristic, a caller that + /// KNOWS the node is final passes . The zero-subscriber guard is the + /// same atomic one in both cases, so a live reader can never lose its subscription to either. + /// + /// True when this call claimed the entry and tore it down. + private bool TryReleaseUnwatched(string path, Lazy lazy, TimeSpan idleWindow, string reason) + { + if (!lazy.IsValueCreated) + return false; + var entry = lazy.Value; + if (!entry.IsIdleCandidate(idleWindow)) + return false; + + var detached = entry.Handle.DetachUpstreams(); + if (!entry.TryMarkIdleEvicted(idleWindow)) + { + // Lost the race — a consumer pinned the entry between the + // pre-check and the mark. Hand the upstreams back; the entry + // keeps serving them. (A later write may open a fresh upstream + // alongside — the same benign divergence a change-feed evict + // produces; both are reaped on the eventual release.) + entry.Handle.ReparkUpstreams(detached); + return false; + } + + _streams.TryRemove(new KeyValuePair>(path, lazy)); + var upstreamReleased = TearDownEntry(path, entry, detached); + logger.LogDebug( + "MeshNodeStreamCache: released {Reason} read stream for {Path} (upstreams disposed: {Count})", + reason, path, detached.Count); + readStreamEvictions.OnNext(new ReadStreamEviction(path, upstreamReleased, reason)); + return true; + } + + /// + public bool ReleaseIfUnwatched(string path) + { + if (System.Threading.Volatile.Read(ref _disposed) != 0) + return false; + if (!_streams.TryGetValue(path, out var lazy)) + return false; + try + { + // TimeSpan.Zero, not a shortened idle window: the CALLER supplied the event that makes + // the entry provably dead (a terminal ActivityLog status), so there is no heuristic left + // to wait out. Everything that protects the sweep still applies verbatim. + return TryReleaseUnwatched(path, lazy, TimeSpan.Zero, "final"); + } + catch (Exception ex) + { + // Releasing a warm cache entry is an optimisation, never a correctness step: the idle + // sweep will get to it. Never fail the write that reported the terminal status. + logger.LogDebug(ex, "MeshNodeStreamCache: releasing final read stream for {Path} failed", path); + return false; + } + } + /// /// Tears down an already-evicted, unlinked entry: disposes the hydration + /// storm-breaker bookkeeping FIRST (so upstream disposal can never surface a diff --git a/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs b/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs index bbc1379749..15c468e85c 100644 --- a/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs +++ b/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs @@ -80,19 +80,76 @@ public static IObservable Append( var options = hub.JsonSerializerOptions; var stream = workspace.GetMeshNodeStream(activityPath); - return stream - .Update(node => - { - // ContentAs, never `is ActivityLog`: a cross-hub update lambda diffs against a LOCAL - // mirror whose Content can be a degraded JsonElement — a plain type test would be - // null, the lambda would no-op, and the patch would never be sent. - if (node.ContentAs(options, logger) is not { } log) return node; - var updated = log.Append(messages); - if (mutate is not null) updated = mutate(updated); - // Claim AFTER the append + mutate so the decision sees the final window size. - return node with { Content = updated.ClaimSeal() }; - }) - .SelectMany(updated => FlushClaimedSeal(hub, activityPath, stream, updated, options, logger)); + // Defer so the settled-node capture below is per SUBSCRIPTION, not shared by every + // subscriber of one cold observable. + return Observable.Defer(() => + { + MeshNode? settled = null; + return stream + .Update(node => + { + // ContentAs, never `is ActivityLog`: a cross-hub update lambda diffs against a LOCAL + // mirror whose Content can be a degraded JsonElement — a plain type test would be + // null, the lambda would no-op, and the patch would never be sent. + if (node.ContentAs(options, logger) is not { } log) return node; + var updated = log.Append(messages); + if (mutate is not null) updated = mutate(updated); + // Claim AFTER the append + mutate so the decision sees the final window size. + return node with { Content = updated.ClaimSeal() }; + }) + .SelectMany(updated => FlushClaimedSeal(hub, activityPath, stream, updated, options, logger)) + // 🚨 onCompleted, NOT onNext. Releasing tears the path's upstream sync streams down, + // and this write is still in flight when its value is emitted — cutting the stream + // out from under it would strand the write's own terminal and leave the per-path + // queue to advance on its 5 s backstop instead. After completion there is nothing + // left to strand, and for a terminal activity there is no next write by definition. + .Do(node => settled = node, + () => + { + if (settled is not null) + ReleaseMirrorWhenFinal(hub, activityPath, settled, options, logger); + }); + }); + } + + /// + /// An activity that has reached a terminal status will never be written again — so the warm + /// mirror the write path opened for it is not a cache, it is a leak with a keep-alive attached. + /// + /// The mechanism, and why "nothing reclaims it" was the wrong diagnosis (#1324). + /// A cross-hub write resolves a shared IMeshNodeStreamCache entry for the path, and that + /// entry's mirror posts a HeartBeatEvent to the owning hub every 45 s for the express + /// purpose of keeping its grain alive. Two reclaimers exist and BOTH are defeated by it: the + /// cache's own idle sweep needs the entry untouched for ten minutes, and Orleans' idle + /// collection (like KernelContainer.DisposeOnTimeout) is re-armed by every inbound + /// message. So a finished compile or import does not merely wait to be collected — it actively + /// prevents its own collection, pinning its _Activity/… node hub and the sync sub-hubs on + /// both sides of the mirror. Measured on NodeTypeRecompileAlcLeakTest, matched runs from + /// an identical baseline: 6.5 → 5.0 retained hubs per compile activity (8.7 → 6.0 per + /// recompile overall), with the mirror-side hubs going 3 → 0. + /// + /// Releasing here is an EVENT, not a shorter timer: the terminal status is proof, supplied + /// by the writer, that the heuristic the sweep is waiting out has already been decided. A reader + /// still watching the finished activity keeps it — ReleaseIfUnwatched makes the same + /// atomic zero-subscriber check the sweep makes, and simply declines when someone is attached. + /// Best-effort by construction: if the release does not happen the idle sweep still gets there, + /// so nothing here may fail the write that reported the status. + /// + private static void ReleaseMirrorWhenFinal( + IMessageHub hub, + string activityPath, + MeshNode final, + System.Text.Json.JsonSerializerOptions options, + ILogger? logger) + { + if (final.ContentAs(options, logger) is not { } log || !log.Status.IsTerminal()) + return; + var cache = hub.ServiceProvider.GetService(); + if (cache is null) + return; + if (cache.ReleaseIfUnwatched(activityPath)) + logger?.LogDebug( + "Activity {Path}: reached {Status} — released its shared mirror", activityPath, log.Status); } /// diff --git a/src/MeshWeaver.Mesh.Contract/Services/IMeshNodeStreamCache.cs b/src/MeshWeaver.Mesh.Contract/Services/IMeshNodeStreamCache.cs index 4b167aa925..bc4121f8f4 100644 --- a/src/MeshWeaver.Mesh.Contract/Services/IMeshNodeStreamCache.cs +++ b/src/MeshWeaver.Mesh.Contract/Services/IMeshNodeStreamCache.cs @@ -104,6 +104,31 @@ IObservable Overwrite( /// void Invalidate(string path); + /// + /// Releases the warm entry for NOW — but ONLY when nothing is reading + /// it. Same protocol and same guard as the idle sweep + /// (MeshNodeStreamCacheOptions.ReadStreamIdleExpiration); the only difference is that the + /// caller supplies an EVENT proving the node is final instead of waiting out the ten-minute + /// heuristic. Returns true when this call claimed the entry and tore it down. + /// + /// Why this exists. The cache keeps a hydrated mirror per path because a path that + /// was just touched is usually about to be touched again — and that mirror posts a + /// HeartBeatEvent to the owner every 45 s expressly to keep its hub/grain alive. For a + /// node with a TERMINAL lifecycle the assumption is provably false: once an + /// ActivityLog reaches a status where IsTerminal() holds, no further write can + /// occur. Leaving the mirror up does not merely postpone reclamation — it actively PREVENTS it, + /// because the heartbeat resets every idle clock the platform has. A finished import or compile + /// therefore pins its own node hub (plus the sync sub-hubs on both sides) for as long as the + /// mirror lives. That is issue #1324's residual. + /// + /// Not a shorter timer. Nothing about the idle heuristic changes; this adds the + /// missing EVENT. A live reader — a user watching the finished activity — still wins: the + /// zero-subscriber check is the same atomic one the sweep makes, so the entry survives and the + /// caller is told false. Idempotent, and a no-op for an unknown path or a path whose node + /// the caller owns (an own-node write never opens a cache entry). + /// + bool ReleaseIfUnwatched(string path); + /// /// Process-wide synced-query cache: returns the cached /// IObservable<IEnumerable<MeshNode>> for the named query, diff --git a/test/MeshWeaver.Graph.Test/IndeterminateProbeReResolutionTest.cs b/test/MeshWeaver.Graph.Test/IndeterminateProbeReResolutionTest.cs index da3dfac42b..da2f9e83cf 100644 --- a/test/MeshWeaver.Graph.Test/IndeterminateProbeReResolutionTest.cs +++ b/test/MeshWeaver.Graph.Test/IndeterminateProbeReResolutionTest.cs @@ -98,6 +98,7 @@ public IObservable Update(string path, Func update public IObservable Overwrite(string path, MeshNode node, JsonSerializerOptions options) => Observable.Never(); public void Invalidate(string path) { } + public bool ReleaseIfUnwatched(string path) => false; public IObservable>? GetQuery(object id) => null; public IObservable> GetQuery(object id, JsonSerializerOptions options, params string[] queries) => Observable.Never>(); diff --git a/test/MeshWeaver.Graph.Test/NodeTypeEnrichmentDoubleCallTest.cs b/test/MeshWeaver.Graph.Test/NodeTypeEnrichmentDoubleCallTest.cs index f12cf32a56..4606ac2d57 100644 --- a/test/MeshWeaver.Graph.Test/NodeTypeEnrichmentDoubleCallTest.cs +++ b/test/MeshWeaver.Graph.Test/NodeTypeEnrichmentDoubleCallTest.cs @@ -80,6 +80,7 @@ public IObservable Update(string path, Func update public IObservable Overwrite(string path, MeshNode node, System.Text.Json.JsonSerializerOptions options) => Observable.Never(); public void Invalidate(string path) { } + public bool ReleaseIfUnwatched(string path) => false; public IObservable>? GetQuery(object id) => null; public IObservable> GetQuery(object id, System.Text.Json.JsonSerializerOptions options, params string[] queries) => Observable.Never>(); diff --git a/test/MeshWeaver.Hosting.Monolith.Test/NodeTypeRecompileAlcLeakTest.cs b/test/MeshWeaver.Hosting.Monolith.Test/NodeTypeRecompileAlcLeakTest.cs index 10d435b06f..cd1814eed5 100644 --- a/test/MeshWeaver.Hosting.Monolith.Test/NodeTypeRecompileAlcLeakTest.cs +++ b/test/MeshWeaver.Hosting.Monolith.Test/NodeTypeRecompileAlcLeakTest.cs @@ -107,30 +107,40 @@ public class NodeTypeRecompileAlcLeakTest(ITestOutputHelper output) : MonolithMe /// — one holder's teardown would kill another's. Measured: compile-state +13 → 0 /// and the NodeType's own hub +3 → 0. /// - /// What still retains, and why the bound is 9 rather than ~2 — one mechanism, and it - /// is a NODE-LIFECYCLE question rather than a stream one. Every compile creates - /// {typePath}/_Activity/compile-<ts> (NodeTypeCompilationHelpers.RunCompile), - /// whose hub is activated by the first activity-log write's SubscribeRequest and which now - /// accounts for essentially the whole residual (+3–5 sync/ apiece, plus the node hub). - /// Nothing in the compile path releases it at terminal status: Complete(...) is the last - /// touch, and there is no hook that drops the mirror, prunes the node, or disposes the hub. + /// What the 8.7 → 6.0 change closed (#1324): the finished activity's mirror was a + /// KEEP-ALIVE, not merely something nobody had collected yet. The previous pass left the + /// question open as "is the residual a permanent leak or a ~25-minute bounded retention", because + /// both existing reclaimers are time-based and invisible to a seconds-long test: the cache's idle + /// sweep needs zero subscribers AND ten minutes untouched + /// (MeshNodeStreamCacheOptions.ReadStreamIdleExpiration), and an idle node hub/grain is + /// collected only after its own idle window. The answer is that the framing was wrong in the + /// consumer's favour: a mirror posts a HeartBeatEvent to its owner every 45 s + /// (SyncStreamOptions.HeartbeatInterval) for the express purpose of keeping the hub alive, + /// and that message re-arms BOTH clocks. A finished compile was not waiting to be reclaimed — it + /// was preventing its own reclamation, for as long as anything kept touching the path. /// - /// 🚨 The two reclamation mechanisms that EXIST are both time-based and therefore invisible - /// to this test, which measures seconds: the shared mesh-node cache's idle sweep needs zero - /// subscribers AND ten minutes untouched (MeshNodeStreamCacheOptions.ReadStreamIdleExpiration), - /// and KernelContainer.DisposeOnTimeout disposes an idle node hub after fifteen — but its - /// timer is reset by EVERY message, and a surviving mirror heart-beats the owner every 45 s - /// (SyncStreamOptions.HeartbeatInterval), which also holds the Orleans grain up via - /// TryDelayDeactivation. So whether the residual is a permanent leak or a ~25-minute - /// bounded retention turns entirely on whether the activity's mirror is actually released once - /// the compile ends — that is the question for the next pass, and it needs a longer-horizon - /// measurement than this one. Do NOT "fix" it by shortening a timer. + /// The cure is an EVENT, not a shorter timer: an ActivityLog that reaches a status + /// where IsTerminal() holds will never be written again, so + /// ActivityLogAppender.Append releases the path's shared entry as part of that same write + /// (IMeshNodeStreamCache.ReleaseIfUnwatched, which makes the SAME atomic zero-subscriber + /// check the sweep makes — a reader still watching the finished activity keeps it). Measured on + /// this repro, matched runs from an identical 38-hub baseline with four retained activities: + /// 6.5 → 5.0 hubs per compile activity, of which the cache/-side mirror hubs go + /// 3 → 0; totals 8.7 → 6.0 per recompile. The direct property — no terminal activity + /// still holds a warm mirror — is asserted at the end of the test, and fails before the fix. + /// + /// What is left, and why the bound is 8 rather than ~1. The whole residual is now the + /// activity's OWN node hub plus the four sync/ sub-hubs its data source builds at startup. + /// Nothing subscribes to it any more, so on Orleans it is finally eligible for the idle + /// deactivation that the heartbeat used to keep pushing away; the monolith has no idle collection + /// for node hubs at all, which is what this test still counts. That is a host-lifecycle question, + /// not a stream one. Do NOT "fix" it by shortening a timer. /// /// So this bound is a RATCHET at the measured value, not an aspiration: it fails the moment - /// the residual gets worse, and it is to be tightened again by whoever closes the activity-hub - /// mechanism above. + /// the residual gets worse — 8 sits above the 6.0/6.7 measured here and below the 8.7 the + /// un-released mirror produced, so a regression of THIS fix reds the test. /// - private const int MaxHubsPerRecompile = 9; + private const int MaxHubsPerRecompile = 8; /// /// Live collectible contexts for this NodeType. The name is @@ -218,6 +228,34 @@ await Mesh.GetWorkspace().GetMeshNodeStream(TypePath) + $"managed = {AfterCollectionBytes() / (1024 * 1024)} MB"); } + var activities = await Mesh.ServiceProvider.GetRequiredService() + .Query(MeshQueryRequest.FromQuery($"namespace:{TypePath}/_Activity")) + .Should().Within(30.Seconds()).Emit(); + + // 🚨 THE MECHANISM, waited on before anything is counted. A terminal activity releases its + // shared mirror as part of the write that reports the terminal status + // (ActivityLogAppender), so this settles as fast as that write lands — but it IS a write, + // and the NodeType flips to Ok slightly ahead of it, so the loop above can return while the + // last compile's Complete(...) is still in flight. Waiting on the CONDITION rather than + // snapshotting into that window is what makes both this assertion and the hub count below + // deterministic; a fixed sleep would race CI either way (WritingTests.md). + // + // `compile-state` is deliberately exempt: it is the type's LIVE compile state, one fixed-id + // node per NodeType rather than one per compile, so it does not grow with recompiles and a + // warm mirror on it is correct. + var compileActivities = (activities?.Items ?? []) + .Select(n => n.Path) + .Where(p => !p.EndsWith($"/{NodeTypeCompileStateMirror.StateId}", StringComparison.Ordinal)) + .ToList(); + var cache = (MeshNodeStreamCache)Mesh.ServiceProvider.GetRequiredService(); + var stillWarm = compileActivities; + await Observable.Interval(50.Milliseconds()).StartWith(0L) + .Select(_ => stillWarm = compileActivities.Where(p => cache.IsReadStreamLive(p)).ToList()) + .Where(warm => warm.Count == 0) + .FirstAsync() + .Timeout(30.Seconds(), Observable.Return(stillWarm)) + .ToTask(); + FullyCollect(); var live = LiveContexts(); var managedGrowth = AfterCollectionBytes() - managedBaseline; @@ -230,15 +268,24 @@ await Mesh.GetWorkspace().GetMeshNodeStream(TypePath) // names its own retainer instead of leaving the next investigation to re-derive it. var hubsAfter = HubAddresses(); var hubGrowth = hubsAfter.Count - hubBaseline.Count; - var activities = await Mesh.ServiceProvider.GetRequiredService() - .Query(MeshQueryRequest.FromQuery($"namespace:{TypePath}/_Activity")) - .Should().Within(30.Seconds()).Emit(); Output.WriteLine( $"WHERE: live hubs {hubBaseline.Count} -> {hubsAfter.Count} (+{hubGrowth} over {Recompiles} " + $"recompiles = {(double)hubGrowth / Recompiles:F1} per recompile), compile-activity nodes " + $"under {TypePath}/_Activity = {activities?.Items?.Count ?? -1}"); - foreach (var line in HubsByParent(hubsAfter.Except(hubBaseline).ToHashSet())) + var newHubs = hubsAfter.Except(hubBaseline).ToHashSet(); + foreach (var line in HubsByParent(newHubs)) Output.WriteLine($" {line}"); + // The SHARP number. The total above is a difference of two whole-mesh counts, so it carries + // the settling of every unrelated framework hub with it — measured 5.3 / 7.0 / 6.3 for the + // SAME code. This one counts only hubs a compile activity is responsible for (its node hub, + // anything hosted under it, and the mirror sync hubs under the shared cache), which is the + // population the retention is about and the one a fix has to move. + var attributed = AttributedToCompileActivities(newHubs); + Output.WriteLine( + $"ATTRIBUTED to compile activities: {attributed.Total} hubs across {attributed.Activities} " + + $"retained activities = {attributed.PerActivity:F1} PER ACTIVITY — " + + $"{attributed.OwnerSide} owner-side (the activity node hub + its own streams), " + + $"{attributed.MirrorSide} mirror-side (the shared cache's client mirror)"); Output.WriteLine($"FINAL live contexts for {TypePath} after {Recompiles} recompiles: {live}"); Output.WriteLine( $"FINAL managed growth over {Recompiles} recompiles: " @@ -297,6 +344,15 @@ await Mesh.GetWorkspace().GetMeshNodeStream(TypePath) + $"({hubGrowth} over {Recompiles}) — each carries an Autofac scope, a TypeRegistry and a " + "JsonSerializerOptions, and in production that is the 130 MB/min curve. The breakdown " + "printed above names which parent hub grew"); + + // 🚨 AND THE MECHANISM, stated directly rather than inferred from a count. Asserted LAST so + // that a failure still carries the full attribution printed above — the diagnosis and the + // verdict arrive in one output. + stillWarm.Should().BeEmpty( + "a terminal activity is never written again, so its shared mirror must be released when " + + "the activity ends — leaving it up does not postpone reclamation, it PREVENTS it: the " + + "mirror posts a HeartBeatEvent to the owner every 45 s expressly to keep its hub/grain " + + $"alive, which re-arms every idle clock the platform has. Still warm: {string.Join(", ", stillWarm)}"); } /// @@ -338,6 +394,53 @@ private IEnumerable HubsByParent(IReadOnlySet newHubs) return counts.OrderByDescending(kv => kv.Value).Select(kv => $"+{kv.Value,3} {kv.Key}"); } + /// + /// Splits the newly-retained hubs into the two halves the compile activities are responsible + /// for, walking the same indented disposal tree does: + /// + /// owner-side — the _Activity/compile-<ts> node hub itself and every + /// hub hosted under it (its own data-source sync streams); + /// mirror-side — the sync/ hubs under the process-wide cache/ hub, + /// i.e. the client end of the shared mirror the write path opened. + /// + /// The mirror side is what the terminal-status release reclaims, and it is what carried the 45 s + /// heartbeat that stopped the owner side from ever going idle. + /// + private (int Total, int OwnerSide, int MirrorSide, int Activities, double PerActivity) + AttributedToCompileActivities(IReadOnlySet newHubs) + { + const string activityMarker = "/_Activity/compile-"; + var owner = 0; + var mirror = 0; + var activities = 0; + var stack = new List(); + foreach (var line in Mesh.GetDisposalDiagnostics().Split('\n')) + { + var m = System.Text.RegularExpressions.Regex.Match(line, @"^(\s*)Hub (\S+) RunLevel"); + if (!m.Success) continue; + var depth = m.Groups[1].Value.Length / 2; + var address = m.Groups[2].Value; + while (stack.Count > depth) stack.RemoveAt(stack.Count - 1); + var parent = stack.Count > 0 ? stack[^1] : ""; + stack.Add(address); + if (!newHubs.Contains(address)) continue; + if (address.Contains(activityMarker, StringComparison.Ordinal)) + { + activities++; + owner++; + } + else if (parent.Contains(activityMarker, StringComparison.Ordinal)) + owner++; + else if (parent.StartsWith("cache/", StringComparison.Ordinal)) + mirror++; + } + // Normalised PER ACTIVITY, because how many of the loop's activities land inside the + // baseline-to-final delta varies run to run (3 or 4) — dividing by Recompiles would fold + // that straight back into the number this metric exists to keep clean. + return (owner + mirror, owner, mirror, activities, + activities == 0 ? 0 : (double)(owner + mirror) / activities); + } + /// /// Managed bytes with a full collection forced FIRST — the opposite choice from /// MemoryDelta, which must never collect because it runs in production. Here the question From dc3f4786b814cb816acc170dd58c0c44476dcde3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Thu, 13 Aug 2026 16:46:39 +0200 Subject: [PATCH 2/2] =?UTF-8?q?review:=20the=20release=20must=20not=20defe?= =?UTF-8?q?r=20Append's=20write=20=E2=80=94=20that=20moved=20the=20identit?= =?UTF-8?q?y=20capture?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI shard 5 caught this and it is the good kind of catch. Wrapping Append's pipeline in Observable.Defer (so the settled-node capture would be per subscription) also deferred the `stream.Update(...)` CALL — and UpdateRemote captures the caller's AccessContext at that call, not at subscribe (`LastModifiedBy = capturedContextAtEntry`). Every activity producer invokes Append inside a scope that has deliberately re-established the activity's OWNER (ActivityRunner + AccessContextScope.FromNode), precisely because the real work runs on pooled hops carrying someone else's ambient identity. Deferring construction moved the capture out of that scope and onto the subscribing thread — the identity-confusion defect ActivityOwnerCredentialTest exists to pin. It failed exactly as designed: `Access denied: user 'activity-intruder' lacks Read permission`. So the Update call goes back to being eager, and `settled` is per Append CALL — which is what this method's contract already implies (an Append is one write, subscribed once), and harmless even under a second subscriber since the release reads only Status. The onCompleted placement, which is about not tearing the upstream down mid-write, is unchanged. MeshWeaver.GitSync.Test 163/163, and the #1324 measurement is unmoved: 5.0 hubs per compile activity, 0 mirror-side, 6.0 per recompile. Co-Authored-By: Claude Opus 5 (1M context) --- .../ActivityLogAppender.cs | 66 +++++++++++-------- 1 file changed, 37 insertions(+), 29 deletions(-) diff --git a/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs b/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs index 15c468e85c..86fa132bca 100644 --- a/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs +++ b/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs @@ -80,36 +80,44 @@ public static IObservable Append( var options = hub.JsonSerializerOptions; var stream = workspace.GetMeshNodeStream(activityPath); - // Defer so the settled-node capture below is per SUBSCRIPTION, not shared by every - // subscriber of one cold observable. - return Observable.Defer(() => - { - MeshNode? settled = null; - return stream - .Update(node => + // 🚨 stream.Update(...) is invoked EAGERLY, here, and it must stay that way — do NOT wrap + // this in Observable.Defer. UpdateRemote captures the caller's AccessContext AT THE .Update() + // CALL, not at subscribe (MeshNodeStreamExtensions: `LastModifiedBy = capturedContextAtEntry`), + // and every activity producer calls Append inside a scope that has deliberately re-established + // the activity's OWNER (ActivityRunner + AccessContextScope.FromNode). Deferring construction + // moves the capture to whatever ambient identity happens to be on the subscribing thread — a + // pooled hop carrying someone else's context in production, which is precisely the + // identity-confusion defect ActivityOwnerCredentialTest exists to pin. It caught this. + // + // `settled` is therefore per Append CALL rather than per subscription, which is what this + // method's contract already implies: an Append is one write, subscribed once. Two subscribers + // to the same cold Append would each re-run the write against the same path, and the release + // reads only Status — so the shared local is harmless even then. + MeshNode? settled = null; + return stream + .Update(node => + { + // ContentAs, never `is ActivityLog`: a cross-hub update lambda diffs against a LOCAL + // mirror whose Content can be a degraded JsonElement — a plain type test would be + // null, the lambda would no-op, and the patch would never be sent. + if (node.ContentAs(options, logger) is not { } log) return node; + var updated = log.Append(messages); + if (mutate is not null) updated = mutate(updated); + // Claim AFTER the append + mutate so the decision sees the final window size. + return node with { Content = updated.ClaimSeal() }; + }) + .SelectMany(updated => FlushClaimedSeal(hub, activityPath, stream, updated, options, logger)) + // 🚨 onCompleted, NOT onNext. Releasing tears the path's upstream sync streams down, and + // this write is still in flight when its value is emitted — cutting the stream out from + // under it would strand the write's own terminal and leave the per-path queue to advance + // on its 5 s backstop instead. After completion there is nothing left to strand, and for + // a terminal activity there is no next write by definition. + .Do(node => settled = node, + () => { - // ContentAs, never `is ActivityLog`: a cross-hub update lambda diffs against a LOCAL - // mirror whose Content can be a degraded JsonElement — a plain type test would be - // null, the lambda would no-op, and the patch would never be sent. - if (node.ContentAs(options, logger) is not { } log) return node; - var updated = log.Append(messages); - if (mutate is not null) updated = mutate(updated); - // Claim AFTER the append + mutate so the decision sees the final window size. - return node with { Content = updated.ClaimSeal() }; - }) - .SelectMany(updated => FlushClaimedSeal(hub, activityPath, stream, updated, options, logger)) - // 🚨 onCompleted, NOT onNext. Releasing tears the path's upstream sync streams down, - // and this write is still in flight when its value is emitted — cutting the stream - // out from under it would strand the write's own terminal and leave the per-path - // queue to advance on its 5 s backstop instead. After completion there is nothing - // left to strand, and for a terminal activity there is no next write by definition. - .Do(node => settled = node, - () => - { - if (settled is not null) - ReleaseMirrorWhenFinal(hub, activityPath, settled, options, logger); - }); - }); + if (settled is not null) + ReleaseMirrorWhenFinal(hub, activityPath, settled, options, logger); + }); } ///