Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
@@ -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.
87 changes: 61 additions & 26 deletions src/MeshWeaver.Hosting/MeshNodeStreamCache.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, Lazy<Entry>>(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)
{
Expand All @@ -1002,6 +977,66 @@ private void ReleaseIdleReadStreams()
}
}

/// <summary>
/// The detach-then-mark release protocol, shared by the timed idle sweep and the event-driven
/// <see cref="ReleaseIfUnwatched"/>. The ONLY difference between the two callers is
/// <paramref name="idleWindow"/> — the sweep waits out the ten-minute heuristic, a caller that
/// KNOWS the node is final passes <see cref="TimeSpan.Zero"/>. The zero-subscriber guard is the
/// same atomic one in both cases, so a live reader can never lose its subscription to either.
/// </summary>
/// <returns>True when this call claimed the entry and tore it down.</returns>
private bool TryReleaseUnwatched(string path, Lazy<Entry> 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<string, Lazy<Entry>>(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;
}

/// <inheritdoc />
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;
}
}

/// <summary>
/// Tears down an already-evicted, unlinked entry: disposes the hydration +
/// storm-breaker bookkeeping FIRST (so upstream disposal can never surface a
Expand Down
67 changes: 66 additions & 1 deletion src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,20 @@ public static IObservable<MeshNode> Append(
var options = hub.JsonSerializerOptions;
var stream = workspace.GetMeshNodeStream(activityPath);

// 🚨 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 =>
{
Expand All @@ -92,7 +106,58 @@ public static IObservable<MeshNode> Append(
// 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));
.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);
});
}

/// <summary>
/// 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.
///
/// <para><b>The mechanism, and why "nothing reclaims it" was the wrong diagnosis (#1324).</b>
/// A cross-hub write resolves a shared <c>IMeshNodeStreamCache</c> entry for the path, and that
/// entry's mirror posts a <c>HeartBeatEvent</c> 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 <c>KernelContainer.DisposeOnTimeout</c>) 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 <c>_Activity/…</c> node hub and the sync sub-hubs on
/// both sides of the mirror. Measured on <c>NodeTypeRecompileAlcLeakTest</c>, matched runs from
/// an identical baseline: <b>6.5 → 5.0 retained hubs per compile activity</b> (8.7 → 6.0 per
/// recompile overall), with the mirror-side hubs going <b>3 → 0</b>.</para>
///
/// <para>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 — <c>ReleaseIfUnwatched</c> 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.</para>
/// </summary>
private static void ReleaseMirrorWhenFinal(
IMessageHub hub,
string activityPath,
MeshNode final,
System.Text.Json.JsonSerializerOptions options,
ILogger? logger)
{
if (final.ContentAs<ActivityLog>(options, logger) is not { } log || !log.Status.IsTerminal())
return;
var cache = hub.ServiceProvider.GetService<IMeshNodeStreamCache>();
if (cache is null)
return;
if (cache.ReleaseIfUnwatched(activityPath))
logger?.LogDebug(
"Activity {Path}: reached {Status} — released its shared mirror", activityPath, log.Status);
}

/// <summary>
Expand Down
25 changes: 25 additions & 0 deletions src/MeshWeaver.Mesh.Contract/Services/IMeshNodeStreamCache.cs
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,31 @@ IObservable<MeshNode> Overwrite(
/// </summary>
void Invalidate(string path);

/// <summary>
/// Releases the warm entry for <paramref name="path"/> NOW — but ONLY when nothing is reading
/// it. Same protocol and same guard as the idle sweep
/// (<c>MeshNodeStreamCacheOptions.ReadStreamIdleExpiration</c>); 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.
///
/// <para><b>Why this exists.</b> 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
/// <c>HeartBeatEvent</c> 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
/// <c>ActivityLog</c> reaches a status where <c>IsTerminal()</c> 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.</para>
///
/// <para><b>Not a shorter timer.</b> 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 <c>false</c>. 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).</para>
/// </summary>
bool ReleaseIfUnwatched(string path);
Comment thread
rbuergi marked this conversation as resolved.

/// <summary>
/// Process-wide synced-query cache: returns the cached
/// <c>IObservable&lt;IEnumerable&lt;MeshNode&gt;&gt;</c> for the named query,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ public IObservable<MeshNode> Update(string path, Func<MeshNode, MeshNode> update
public IObservable<MeshNode> Overwrite(string path, MeshNode node, JsonSerializerOptions options)
=> Observable.Never<MeshNode>();
public void Invalidate(string path) { }
public bool ReleaseIfUnwatched(string path) => false;
public IObservable<IEnumerable<MeshNode>>? GetQuery(object id) => null;
public IObservable<IEnumerable<MeshNode>> GetQuery(object id, JsonSerializerOptions options, params string[] queries)
=> Observable.Never<IEnumerable<MeshNode>>();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ public IObservable<MeshNode> Update(string path, Func<MeshNode, MeshNode> update
public IObservable<MeshNode> Overwrite(string path, MeshNode node, System.Text.Json.JsonSerializerOptions options)
=> Observable.Never<MeshNode>();
public void Invalidate(string path) { }
public bool ReleaseIfUnwatched(string path) => false;
public IObservable<IEnumerable<MeshNode>>? GetQuery(object id) => null;
public IObservable<IEnumerable<MeshNode>> GetQuery(object id, System.Text.Json.JsonSerializerOptions options, params string[] queries)
=> Observable.Never<IEnumerable<MeshNode>>();
Expand Down
Loading
Loading