From 33b4d156e00c29b88733f3aa5b82d0c63743b9cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:45:15 +0200 Subject: [PATCH 1/3] Kernel: the idle-disconnect never disposes a hub whose submission is still running (#4422) KernelContainer.DisposeOnTimeout re-armed its one-shot 15-minute timer only on messages DELIVERED to the kernel-hosting hub. A running script delivers none: it executes on the hosted kernelExec child, its mesh reads answer elsewhere, and its log lines leave as outgoing writes. So a script still working 15 min after the last inbound message had the hub, and its executor, disposed mid-run with no SubmitCodeResponse and no terminal status. Measured on memex.systemorph.com 2026-09-15: an approved OperationRequest's follow loop went silent at 10:45:06Z and the request's 20-minute silence watchdog failed it at 11:05:06Z. - IdleState (weak hub ref, window, in-flight count, weak executor ref): the forwarder Begins on every SubmitCodeRequest and Ends in Finally, so a response, an error or a torn-down subscription all release it. - Elapsed re-arms while a submission is in flight on a LIVE executor and disposes otherwise, so a dead executor cannot pin the hub. - End restarts the window, so a finished hub is reclaimed one full window after its run ends, which keeps the #1324/#1435 reclamation. - KernelIdleDisconnectWhileRunningTest pins both directions. Co-Authored-By: Claude Opus 5 (1M context) --- src/MeshWeaver.Kernel.Hub/KernelContainer.cs | 92 +++++++++++++++++-- src/MeshWeaver.Kernel/KernelHubOptions.cs | 5 + .../KernelIdleDisconnectWhileRunningTest.cs | 85 +++++++++++++++++ 3 files changed, 174 insertions(+), 8 deletions(-) create mode 100644 test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs diff --git a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs index 2e191364a6..3df52b19c3 100644 --- a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs +++ b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs @@ -40,6 +40,12 @@ public class KernelContainer(IServiceProvider serviceProvider) private IMessageHub? executorHub; private readonly object executorHubLock = new(); + /// + /// The idle-disconnect's shared state β€” set by , read by the + /// submission forwarder. Holds the hub only WEAKLY (see ). + /// + private IdleState? idle; + /// /// Idle window after which a kernel-hosting hub disposes itself. 15 minutes by default; a host /// (or a test that needs to observe the reclamation inside its budget) can override it by @@ -194,23 +200,86 @@ private void DisposeOnTimeout(IMessageHub hub) // [RunLevel=1] β€” the MeshHub_IsCollected leak). A WeakReference + static callback // lets the GC reclaim an unreferenced hub: if it's already collectable the // disconnect is moot; if it's still alive the weak ref resolves and disposes it. - var weakHub = new WeakReference(hub); - var timer = new Timer(static state => - { - if (((WeakReference)state!).TryGetTarget(out var h)) - h.Dispose(); - }, weakHub, disconnectTimeout, Timeout.InfiniteTimeSpan); + // + // 🚨 IDLE MEANS "NO MESSAGE *AND* NOTHING EXECUTING" (MeshWeaver#4422). The timer is + // re-armed by messages DELIVERED to this hub, and a running script delivers none: it + // executes on the hosted executor, its mesh reads answer elsewhere, and its log lines + // leave as outgoing writes. So a script still working 15 min after the last inbound + // message had this hub β€” and its executor with it β€” disposed mid-run, silently: no + // SubmitCodeResponse, no terminal status (an approved OperationRequest died that way on + // memex, 2026-09-15). While a forwarded submission is in flight the callback re-arms + // instead of disposing; the finished-activity reclamation (#1324/#1435) is unchanged, + // because a finished run has nothing in flight and the window restarts when it ends. + var state = new IdleState(new WeakReference(hub), disconnectTimeout); + var timer = new Timer(static s => ((IdleState)s!).Elapsed(), state, disconnectTimeout, Timeout.InfiniteTimeSpan); + state.Timer = timer; + idle = state; // Dispose the timer WITH the hub for the normal (activated β†’ disposed) path so // the queue drops it promptly rather than waiting on a GC. hub.RegisterForDisposal((IDisposable)timer); - var idle = disconnectTimeout; hub.Register(d => { - timer.Change(idle, Timeout.InfiniteTimeSpan); + state.Rearm(); return d; }); } + /// + /// The idle-disconnect's state: the hub (WEAKLY β€” the timer queue is a GC root), the window, + /// and how many forwarded submissions have not answered yet. The timer holds this object and + /// this object holds the timer β€” a cycle the GC collects; nothing here roots the hub. + /// + internal sealed class IdleState(WeakReference hub, TimeSpan window) + { + private int inFlight; + private WeakReference? executor; + + /// The one-shot timer this state re-arms. + public Timer? Timer { get; set; } + + /// Submissions forwarded to the executor that have not answered yet. + public int InFlight => Volatile.Read(ref inFlight); + + /// Remembers the executor so a dead one cannot keep the hub alive forever. + public void Executor(IMessageHub hosted) => executor = new WeakReference(hosted); + + /// A submission was forwarded β€” the hub is executing, not idle. + public void Begin() => Interlocked.Increment(ref inFlight); + + /// A submission answered (or failed) β€” the idle window starts again from now. + public void End() + { + if (Interlocked.Decrement(ref inFlight) < 0) + Interlocked.Exchange(ref inFlight, 0); + Rearm(); + } + + /// Restart the idle window. + public void Rearm() + { + try { Timer?.Change(window, Timeout.InfiniteTimeSpan); } + catch (ObjectDisposedException) { /* the hub is gone; nothing to keep alive */ } + } + + /// + /// The window elapsed. A submission still executing on a LIVE executor keeps the hub; a + /// dead executor cannot answer, so its count no longer means "working" and the hub goes. + /// + public void Elapsed() + { + if (InFlight > 0 && ExecutorAlive()) + { + Rearm(); + return; + } + if (hub.TryGetTarget(out var h)) + h.Dispose(); + } + + private bool ExecutorAlive() => + executor is not null && executor.TryGetTarget(out var e) && !e.IsDisposing; + } + ISynchronizationStream> GetAreaStream(IServiceProvider sp) => sp.GetRequiredService>>(); @@ -255,6 +324,7 @@ private IMessageHub GetOrCreateExecutor(IMessageHub publicHub) var executor = new KernelExecutor(publicHub); executorHub = publicHub.GetHostedHub(execAddress, executor.Configure, HostedHubCreation.Always) ?? throw new InvalidOperationException($"Failed to create kernel executor hub at {execAddress}"); + idle?.Executor(executorHub); return executorHub; } } @@ -268,8 +338,14 @@ private IMessageDelivery ForwardSubmitCodeRequest(IMessageHub hub, IMessageDeliv { var executor = GetOrCreateExecutor(hub); + // In flight until the executor answers β€” the idle-disconnect must not dispose this hub + // under a running script (MeshWeaver#4422). Finally runs on the response, on an error, + // and on a disposal that tears the subscription down, so the count cannot leak upward. + var state = idle; + state?.Begin(); hub.Observe(request.Message, o => o.WithTarget(executor.Address)) .Take(1) + .Finally(() => state?.End()) .Subscribe( resp => hub.Post(resp.Message, o => o.ResponseFor(request)), ex => hub.Post( diff --git a/src/MeshWeaver.Kernel/KernelHubOptions.cs b/src/MeshWeaver.Kernel/KernelHubOptions.cs index bcab445964..9a7d406308 100644 --- a/src/MeshWeaver.Kernel/KernelHubOptions.cs +++ b/src/MeshWeaver.Kernel/KernelHubOptions.cs @@ -26,6 +26,11 @@ public sealed record KernelHubOptions /// settable so a TEST can assert the reclamation property inside a test budget /// (CompileActivityHubRetentionTest), and for a host that genuinely wants a different /// idle policy β€” not as a memory tuning knob. Default: 15 minutes. + /// 🚨 Idle means no inbound message AND no submission in flight (#4422). The + /// window is re-armed by delivered messages, and a running script delivers none, so without the + /// in-flight check a script still working 15 min after the last inbound message had its hub + /// disposed mid-run, silently. KernelContainer.IdleState keeps the hub while a forwarded + /// submission has not answered and restarts the window when it does. /// public TimeSpan IdleDisconnectTimeout { get; init; } = TimeSpan.FromMinutes(15); } diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs new file mode 100644 index 0000000000..253c39307e --- /dev/null +++ b/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs @@ -0,0 +1,85 @@ +using System; +using MeshWeaver.Kernel.Hub; +using MeshWeaver.Messaging; +using NSubstitute; +using Xunit; + +namespace MeshWeaver.Compiler.Pipeline.Test; + +/// +/// 🚨 The kernel's idle-disconnect must not dispose a hub whose submission is still executing +/// (MeshWeaver#4422). The timer is re-armed only by messages DELIVERED to the hub, and a running +/// script delivers none β€” so on memex (2026-09-15) an approved OperationRequest's script died +/// silently 15 min after the last inbound message, mid-run. Idle now means "no message AND nothing +/// in flight", and the reclamation of a FINISHED hub (#1324/#1435) is unchanged β€” pinned in both +/// directions below. +/// +public class KernelIdleDisconnectWhileRunningTest +{ + private static (KernelContainer.IdleState State, IMessageHub Hub, IMessageHub Executor) Arrange() + { + var hub = Substitute.For(); + var executor = Substitute.For(); + executor.IsDisposing.Returns(false); + var state = new KernelContainer.IdleState(new WeakReference(hub), TimeSpan.FromMinutes(15)); + state.Executor(executor); + return (state, hub, executor); + } + + [Fact] + public void RunningSubmission_KeepsTheHub() + { + var (state, hub, _) = Arrange(); + state.Begin(); + + state.Elapsed(); + + hub.DidNotReceive().Dispose(); + Assert.Equal(1, state.InFlight); + } + + [Fact] + public void FinishedSubmission_IsReclaimedAsBefore() + { + var (state, hub, _) = Arrange(); + state.Begin(); + state.End(); + + state.Elapsed(); + + hub.Received(1).Dispose(); + } + + [Fact] + public void NothingEverSubmitted_IsReclaimedAsBefore() + { + var (state, hub, _) = Arrange(); + + state.Elapsed(); + + hub.Received(1).Dispose(); + } + + [Fact] + public void DeadExecutor_CannotKeepTheHubAliveForever() + { + var (state, hub, executor) = Arrange(); + state.Begin(); + executor.IsDisposing.Returns(true); + + state.Elapsed(); + + hub.Received(1).Dispose(); + } + + [Fact] + public void UnbalancedEnd_NeverCountsBelowZero() + { + var (state, _, _) = Arrange(); + state.End(); + state.End(); + state.Begin(); + + Assert.Equal(1, state.InFlight); + } +} From 5846d9cc3568b7221e7e213a24416300efd0a384 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Tue, 15 Sep 2026 19:27:01 +0200 Subject: [PATCH 2/3] Kernel idle-disconnect: claims are one atomic snapshot, taken before setup, released on every path (#4422) Answers the six #4423 review threads. - One immutable Snapshot(Closed, ImmutableHashSet) swapped with ImmutableInterlocked.Update replaces the counter plus clamp. A claim, a release and the close are each one CAS on the same reference, so a claim is ordered against the close. Nothing can go negative, and a concurrent claim can no longer be erased by a clamp. - IdleState.Track takes the claim BEFORE the executor is set up. A claim with no executor bound yet counts as working, so a tick during setup keeps the hub. A claim after the close is refused and nothing is set up. - The setup and the post run inside Observable.Defer(...).Finally(Release). A synchronous throw from Observe's post or from GetOrCreateExecutor releases the claim instead of pinning the hub. - The timer comes from KernelHubOptions.TimeProvider (TimeProvider.System in every host). KernelIdleDisconnectWhileRunningTest now uses real HubTestBase hubs, not NSubstitute, and steps a manual clock through the production window. It asserts the due time moves on response, error and teardown, and that a dead executor cannot pin the hub. - What's New entry (Category: Fix), and the in-flight rule in Doc/Architecture/MeshNodeStreamCache. Co-Authored-By: Claude Opus 5 (1M context) --- .../Data/Architecture/MeshNodeStreamCache.md | 13 ++ ...o-longer-stopped-after-15-quiet-minutes.md | 38 ++++ src/MeshWeaver.Kernel.Hub/KernelContainer.cs | 156 +++++++++---- src/MeshWeaver.Kernel/KernelHubOptions.cs | 9 +- .../KernelIdleDisconnectWhileRunningTest.cs | 205 ++++++++++++++---- 5 files changed, 331 insertions(+), 90 deletions(-) create mode 100644 src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-15-a-long-running-script-is-no-longer-stopped-after-15-quiet-minutes.md diff --git a/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md b/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md index 9e2262f60f..4af0c0fa68 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md @@ -311,6 +311,19 @@ that can only pass is not a guard. lesson of this section is that the clocks were correct and something was resetting them; a shorter window would have hidden that defect rather than cured it. +🚨 **Idle means no inbound message AND nothing in flight** (#4422). A running script delivers no +message to the hub that hosts its kernel: it executes on the hosted `kernelExec/…` child, and its +progress lines leave as outgoing writes. Re-armed only by inbound messages, the window therefore +disposed a hub under a script still working 15 minutes after its last input, silently. That is how an +approved OperationRequest died on memex on 2026-09-15. `KernelContainer.IdleState` now holds each +forwarded `SubmitCodeRequest` as a **claim**, taken before the executor is set up and released on the +response, an error or a teardown. On the tick it keeps the hub while any claim is still being set up +or sits on a live executor, and re-arms. A released claim restarts the window, so the reclamation +above is unchanged, one window after the last run. The claims and the close decision are one +immutable snapshot swapped atomically, so a claim and a close are always ordered. +`KernelIdleDisconnectWhileRunningTest` steps through the production window on an injected clock +(`KernelHubOptions.TimeProvider`). + ### 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-09-15-a-long-running-script-is-no-longer-stopped-after-15-quiet-minutes.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-15-a-long-running-script-is-no-longer-stopped-after-15-quiet-minutes.md new file mode 100644 index 0000000000..45a49f1f76 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-15-a-long-running-script-is-no-longer-stopped-after-15-quiet-minutes.md @@ -0,0 +1,38 @@ +--- +Name: A long-running script is no longer stopped after 15 quiet minutes +Category: Fix +Description: A script that was still working 15 minutes after anything was last sent to it was stopped mid-run, silently β€” no result, no error, and the run never finished. Scripts now run for as long as they work, and an idle kernel is still cleaned up 15 minutes after its last run ends. +Icon: Timer +Order: -20260915 +--- + +# A long-running script is no longer stopped after 15 quiet minutes + +A script that ran for a long time could simply **stop**. Not fail: stop. Its output stopped growing, +no error appeared, and the run never reached *Done* or *Failed*. Anything waiting on it waited until +its own watchdog gave up. An approved operation request did exactly that on 2026-09-15: it reported +its last line at 10:45, and nothing after. + +The cause was the kernel's idle clean-up. A kernel that nobody uses is closed after 15 minutes, so an +abandoned one does not hold memory for ever. "Nobody uses it" was measured as "nothing has been sent +to it for 15 minutes". A running script sends the kernel nothing, though. It works on its own, and +its progress lines travel the other way. So a script still busy 15 minutes after its last input +looked idle, and the clean-up closed the kernel under it. + +## What changed + +**Idle now means "nothing sent to it *and* nothing running".** While a submitted script has not +answered, the clean-up keeps the kernel and checks again one window later. When the script finishes, +fails or is stopped, the 15 minutes start again from that moment. + +Two limits stay in place, so a kernel cannot be kept forever: + +- **A finished kernel is still closed 15 minutes after its last run.** That behaviour is unchanged. +- **A kernel whose executor has died is closed.** A dead executor can never answer, so the script it + was running no longer counts as working. + +## What you will see + +Long scripts, such as a migration, a partition drain or a provisioning run, now finish or fail with +their own result. They no longer go quiet part-way through. Nothing needs to be re-installed: the fix +is in the platform and arrives with the next update of your deployment. diff --git a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs index 3df52b19c3..b482967656 100644 --- a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs +++ b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs @@ -47,13 +47,13 @@ public class KernelContainer(IServiceProvider serviceProvider) private IdleState? idle; /// - /// Idle window after which a kernel-hosting hub disposes itself. 15 minutes by default; a host - /// (or a test that needs to observe the reclamation inside its budget) can override it by - /// registering a singleton. 🚨 NOT a memory tuning knob β€” see - /// . + /// The idle window after which a kernel-hosting hub disposes itself (15 minutes by default) and + /// the clock it runs on. A host (or a test that needs to observe the reclamation inside its + /// budget) can override either by registering a singleton. + /// 🚨 NOT a memory tuning knob β€” see . /// - private readonly TimeSpan disconnectTimeout = - (serviceProvider.GetService() ?? new KernelHubOptions()).IdleDisconnectTimeout; + private readonly KernelHubOptions options = + serviceProvider.GetService() ?? new KernelHubOptions(); /// /// Hub configuration for the standalone kernel-hub (full mesh-types + routes). @@ -210,13 +210,11 @@ private void DisposeOnTimeout(IMessageHub hub) // memex, 2026-09-15). While a forwarded submission is in flight the callback re-arms // instead of disposing; the finished-activity reclamation (#1324/#1435) is unchanged, // because a finished run has nothing in flight and the window restarts when it ends. - var state = new IdleState(new WeakReference(hub), disconnectTimeout); - var timer = new Timer(static s => ((IdleState)s!).Elapsed(), state, disconnectTimeout, Timeout.InfiniteTimeSpan); - state.Timer = timer; + var state = new IdleState(new WeakReference(hub), options.IdleDisconnectTimeout, options.TimeProvider); idle = state; // Dispose the timer WITH the hub for the normal (activated β†’ disposed) path so // the queue drops it promptly rather than waiting on a GC. - hub.RegisterForDisposal((IDisposable)timer); + hub.RegisterForDisposal(state); hub.Register(d => { state.Rearm(); @@ -225,59 +223,114 @@ private void DisposeOnTimeout(IMessageHub hub) } /// - /// The idle-disconnect's state: the hub (WEAKLY β€” the timer queue is a GC root), the window, - /// and how many forwarded submissions have not answered yet. The timer holds this object and - /// this object holds the timer β€” a cycle the GC collects; nothing here roots the hub. + /// The idle-disconnect: the hub (WEAKLY β€” the timer queue is a GC root), the window, its one-shot + /// timer, and the submissions forwarded to the executor that have not answered yet. The timer + /// holds this object and this object holds the timer β€” a cycle the GC collects; nothing here + /// roots the hub. + /// + /// 🚨 ONE atomic snapshot, never a counter beside a flag (the #4423 review). The + /// timer's decision β€” "nothing is working, so close and dispose" β€” and a submission's β€” "claim a + /// slot unless closed" β€” are each ONE compare-and-swap on the same immutable + /// , so the two are totally ordered: a claim lands before the close (and + /// the close does not happen) or after it (and the claim is refused; the hub is going). There is + /// no count to go negative, and no window between a decrement and a clamp for a concurrent + /// claim to fall into: a submission's claim is an OBJECT in a set, removed by that claim alone, + /// and removing it twice is a no-op. /// - internal sealed class IdleState(WeakReference hub, TimeSpan window) + internal sealed class IdleState : IDisposable { - private int inFlight; - private WeakReference? executor; + private readonly WeakReference hub; + private readonly TimeSpan window; + private readonly ITimer timer; + private Snapshot current = new(false, ImmutableHashSet.Empty); - /// The one-shot timer this state re-arms. - public Timer? Timer { get; set; } + /// Starts the window on β€” in every host. + public IdleState(WeakReference hub, TimeSpan window, TimeProvider clock) + { + this.hub = hub; + this.window = window; + // A STATIC callback over this state: the hub is reachable from the timer only weakly. + timer = clock.CreateTimer(static s => ((IdleState)s!).Elapsed(), this, window, Timeout.InfiniteTimeSpan); + } /// Submissions forwarded to the executor that have not answered yet. - public int InFlight => Volatile.Read(ref inFlight); + public int InFlight => Volatile.Read(ref current).InFlight.Count; - /// Remembers the executor so a dead one cannot keep the hub alive forever. - public void Executor(IMessageHub hosted) => executor = new WeakReference(hosted); + /// The window elapsed with nothing working, or the hub was disposed: no claim is accepted any more. + public bool Closed => Volatile.Read(ref current).Closed; - /// A submission was forwarded β€” the hub is executing, not idle. - public void Begin() => Interlocked.Increment(ref inFlight); + /// + /// Runs one submission as IN FLIGHT: the claim is taken on subscribe, BEFORE + /// sets anything up, so a window that elapses during the setup + /// sees a submission being set up rather than an idle hub. runs + /// INSIDE the tracked observable, so a synchronous throw from the setup or from the post + /// reaches Finally like a response, an error or a teardown does β€” every path releases + /// the claim, and none can pin the hub for ever. A hub the timer already closed refuses the + /// claim and never runs . + /// + public IObservable Track(Func> dispatch) => + Observable.Defer(() => + { + var claim = new Claim(); + if (!ImmutableInterlocked.Update(ref current, s => s.Closed ? s : s with { InFlight = s.InFlight.Add(claim) })) + return Observable.Throw(new ObjectDisposedException(nameof(KernelContainer), + "The kernel host was reclaimed after being idle; submit the code again.")); + return Observable.Defer(() => dispatch(claim)).Finally(() => Release(claim)); + }); - /// A submission answered (or failed) β€” the idle window starts again from now. - public void End() + /// A submission answered, failed or was torn down β€” the idle window starts again from now. + private void Release(Claim claim) { - if (Interlocked.Decrement(ref inFlight) < 0) - Interlocked.Exchange(ref inFlight, 0); + ImmutableInterlocked.Update(ref current, s => s.InFlight.Contains(claim) ? s with { InFlight = s.InFlight.Remove(claim) } : s); Rearm(); } /// Restart the idle window. public void Rearm() { - try { Timer?.Change(window, Timeout.InfiniteTimeSpan); } - catch (ObjectDisposedException) { /* the hub is gone; nothing to keep alive */ } + if (Volatile.Read(ref current).Closed) + return; + try { timer.Change(window, Timeout.InfiniteTimeSpan); } + catch (ObjectDisposedException) { /* disposed with the hub between the read above and here */ } } /// - /// The window elapsed. A submission still executing on a LIVE executor keeps the hub; a - /// dead executor cannot answer, so its count no longer means "working" and the hub goes. + /// The window elapsed. A submission still executing on a LIVE executor (or still being set + /// up) keeps the hub and re-arms the window; a dead executor cannot answer, so its claim no + /// longer means "working", and a hub whose claims are all dead goes. /// - public void Elapsed() + internal void Elapsed() { - if (InFlight > 0 && ExecutorAlive()) + if (ImmutableInterlocked.Update(ref current, s => s.Closed || s.InFlight.Any(c => c.Working) ? s : s with { Closed = true })) { - Rearm(); + if (hub.TryGetTarget(out var h)) + h.Dispose(); return; } - if (hub.TryGetTarget(out var h)) - h.Dispose(); + Rearm(); + } + + /// The hub is going: refuse further claims and drop the timer from the queue. + public void Dispose() + { + ImmutableInterlocked.Update(ref current, s => s.Closed ? s : s with { Closed = true }); + timer.Dispose(); } - private bool ExecutorAlive() => - executor is not null && executor.TryGetTarget(out var e) && !e.IsDisposing; + private sealed record Snapshot(bool Closed, ImmutableHashSet InFlight); + + /// One forwarded submission's slot, bound to the executor it was sent to once that exists. + internal sealed class Claim + { + private WeakReference? executor; + + /// The submission was sent to β€” only a live one can still answer. + public void Bind(IMessageHub hosted) => Volatile.Write(ref executor, new WeakReference(hosted)); + + /// Still being set up (no executor yet), or sent to an executor that is alive. + internal bool Working => + Volatile.Read(ref executor) is not { } bound || (bound.TryGetTarget(out var e) && !e.IsDisposing); + } } ISynchronizationStream> GetAreaStream(IServiceProvider sp) @@ -324,7 +377,6 @@ private IMessageHub GetOrCreateExecutor(IMessageHub publicHub) var executor = new KernelExecutor(publicHub); executorHub = publicHub.GetHostedHub(execAddress, executor.Configure, HostedHubCreation.Always) ?? throw new InvalidOperationException($"Failed to create kernel executor hub at {execAddress}"); - idle?.Executor(executorHub); return executorHub; } } @@ -336,16 +388,15 @@ private IMessageHub GetOrCreateExecutor(IMessageHub publicHub) /// private IMessageDelivery ForwardSubmitCodeRequest(IMessageHub hub, IMessageDelivery request) { - var executor = GetOrCreateExecutor(hub); - - // In flight until the executor answers β€” the idle-disconnect must not dispose this hub - // under a running script (MeshWeaver#4422). Finally runs on the response, on an error, - // and on a disposal that tears the subscription down, so the count cannot leak upward. - var state = idle; - state?.Begin(); - hub.Observe(request.Message, o => o.WithTarget(executor.Address)) + // In flight from BEFORE the executor is set up until it answers β€” the idle-disconnect must + // not dispose this hub under a running script (MeshWeaver#4422). The executor setup AND the + // post run inside the tracked observable: Observe posts before it returns and rethrows a + // synchronous post failure, so outside it a throw would skip Finally and leak the claim. + var submission = idle is { } state + ? state.Track(claim => Dispatch(hub, request.Message, claim)) + : Observable.Defer(() => Dispatch(hub, request.Message, null)); + submission .Take(1) - .Finally(() => state?.End()) .Subscribe( resp => hub.Post(resp.Message, o => o.ResponseFor(request)), ex => hub.Post( @@ -355,6 +406,15 @@ private IMessageDelivery ForwardSubmitCodeRequest(IMessageHub hub, IMessageDeliv return request.Processed(); } + /// Sets up (or reuses) the executor, binds the claim to it, and posts the submission. + private IObservable> Dispatch( + IMessageHub hub, SubmitCodeRequest submission, IdleState.Claim? claim) + { + var executor = GetOrCreateExecutor(hub); + claim?.Bind(executor); + return hub.Observe(submission, o => o.WithTarget(executor.Address)); + } + /// /// Forward to the executor (if it has been /// materialised β€” no point spawning the executor just to cancel nothing). diff --git a/src/MeshWeaver.Kernel/KernelHubOptions.cs b/src/MeshWeaver.Kernel/KernelHubOptions.cs index 9a7d406308..0ef683179d 100644 --- a/src/MeshWeaver.Kernel/KernelHubOptions.cs +++ b/src/MeshWeaver.Kernel/KernelHubOptions.cs @@ -30,7 +30,14 @@ public sealed record KernelHubOptions /// window is re-armed by delivered messages, and a running script delivers none, so without the /// in-flight check a script still working 15 min after the last inbound message had its hub /// disposed mid-run, silently. KernelContainer.IdleState keeps the hub while a forwarded - /// submission has not answered and restarts the window when it does. + /// submission is being set up or has not answered, and restarts the window when it does. /// public TimeSpan IdleDisconnectTimeout { get; init; } = TimeSpan.FromMinutes(15); + + /// + /// The clock runs on β€” in + /// every host. A TEST injects its own so it can step through the window deterministically + /// instead of waiting it out or shrinking it (KernelIdleDisconnectWhileRunningTest). + /// + public TimeProvider TimeProvider { get; init; } = TimeProvider.System; } diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs index 253c39307e..3b30e8c7c5 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs @@ -1,7 +1,15 @@ using System; +using System.Collections.Immutable; +using System.Linq; +using System.Reactive; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Fixture; +using MeshWeaver.Kernel; using MeshWeaver.Kernel.Hub; using MeshWeaver.Messaging; -using NSubstitute; using Xunit; namespace MeshWeaver.Compiler.Pipeline.Test; @@ -13,73 +21,188 @@ namespace MeshWeaver.Compiler.Pipeline.Test; /// silently 15 min after the last inbound message, mid-run. Idle now means "no message AND nothing /// in flight", and the reclamation of a FINISHED hub (#1324/#1435) is unchanged β€” pinned in both /// directions below. +/// +/// Real hubs, an injected clock. The kernel host and its executor are real hosted hubs, so +/// "disposed" is the hub's own and a dead executor is one +/// that was really disposed. The window runs on a β€” the +/// seam β€” so every assertion is about WHEN the one-shot +/// timer fires, stepped deterministically through the production window rather than waited out +/// or shrunk. /// -public class KernelIdleDisconnectWhileRunningTest +public class KernelIdleDisconnectWhileRunningTest(ITestOutputHelper output) : HubTestBase(output) { - private static (KernelContainer.IdleState State, IMessageHub Hub, IMessageHub Executor) Arrange() + /// The production window β€” the test steps through it, never shortens it. + private static readonly TimeSpan Window = new KernelHubOptions().IdleDisconnectTimeout; + + /// How the executor's answer ends a submission β€” every one must restart the window. + public enum Termination { Response, Error, Teardown } + + [Theory] + [InlineData(Termination.Response)] + [InlineData(Termination.Error)] + [InlineData(Termination.Teardown)] + public async Task ARunningSubmission_OutlivesTheWindow_AndTheWindowRestartsWhenItEnds(Termination end) { - var hub = Substitute.For(); - var executor = Substitute.For(); - executor.IsDisposing.Returns(false); - var state = new KernelContainer.IdleState(new WeakReference(hub), TimeSpan.FromMinutes(15)); - state.Executor(executor); - return (state, hub, executor); + var (kernel, executor, idle, clock) = await ArrangeAsync(); + var answer = new Subject(); + var subscription = idle.Track(claim => { claim.Bind(executor); return answer; }) + .Take(1) + .Subscribe(_ => { }, _ => { }); + + clock.Advance(Window); + Assert.False(kernel.IsDisposing, "a submission still executing on a live executor keeps its hub past the window"); + Assert.Equal(1, idle.InFlight); + + clock.Advance(Window / 2); + switch (end) + { + case Termination.Response: answer.OnNext(Unit.Default); break; + case Termination.Error: answer.OnError(new InvalidOperationException("the script failed")); break; + case Termination.Teardown: subscription.Dispose(); break; + } + Assert.Equal(0, idle.InFlight); + + // The keep at the first tick re-armed the timer for 2Γ—Window. Ending the run half a window + // later must have moved it to 2.5Γ—Window β€” otherwise the hub goes half a window after its run. + clock.Advance(Window / 2); + Assert.False(kernel.IsDisposing, $"the window restarts when the run ends ({end}), so the hub is not reclaimed at the earlier due time"); + + clock.Advance(Window / 2); + Assert.True(kernel.IsDisposing, "a finished hub is still reclaimed one full window after its run (#1324/#1435)"); } [Fact] - public void RunningSubmission_KeepsTheHub() + public async Task ADeadExecutor_CannotKeepTheHubAliveForever() { - var (state, hub, _) = Arrange(); - state.Begin(); + var (kernel, executor, idle, clock) = await ArrangeAsync(); + idle.Track(claim => { claim.Bind(executor); return Observable.Never(); }).Subscribe(); - state.Elapsed(); + clock.Advance(Window); + Assert.False(kernel.IsDisposing, "kept while the executor is alive"); - hub.DidNotReceive().Dispose(); - Assert.Equal(1, state.InFlight); + executor.Dispose(); + // The keep above re-armed the one-shot timer; without that re-arm nothing would ever fire again. + clock.Advance(Window); + Assert.True(kernel.IsDisposing, "a dead executor cannot answer, so its submission no longer holds the hub"); + Assert.True(idle.Closed); } [Fact] - public void FinishedSubmission_IsReclaimedAsBefore() + public async Task TheWindowElapsingDuringExecutorSetup_KeepsTheHub() { - var (state, hub, _) = Arrange(); - state.Begin(); - state.End(); - - state.Elapsed(); - - hub.Received(1).Dispose(); + var (kernel, executor, idle, clock) = await ArrangeAsync(); + + idle.Track(claim => + { + // The timer fires while the forwarder is still creating the executor. + clock.Advance(Window); + claim.Bind(executor); + return Observable.Never(); + }).Subscribe(); + + Assert.False(kernel.IsDisposing, "the claim is taken BEFORE the setup, so the tick sees a submission being set up"); + Assert.Equal(1, idle.InFlight); } [Fact] - public void NothingEverSubmitted_IsReclaimedAsBefore() + public async Task ASetupThatThrows_ReleasesItsClaim() { - var (state, hub, _) = Arrange(); + var (kernel, _, idle, clock) = await ArrangeAsync(); + Exception? failure = null; - state.Elapsed(); + // What IMessageHub.Observe does when the post itself fails: it throws before returning. + idle.Track(_ => throw new InvalidOperationException("the post was refused")) + .Subscribe(_ => { }, ex => failure = ex); - hub.Received(1).Dispose(); + Assert.IsType(failure); + Assert.Equal(0, idle.InFlight); + clock.Advance(Window); + Assert.True(kernel.IsDisposing, "a submission whose setup threw holds nothing, so the idle hub is reclaimed"); } [Fact] - public void DeadExecutor_CannotKeepTheHubAliveForever() + public async Task AClaimAfterTheHubClosed_IsRefused_AndNothingIsSetUp() { - var (state, hub, executor) = Arrange(); - state.Begin(); - executor.IsDisposing.Returns(true); - - state.Elapsed(); + var (kernel, _, idle, clock) = await ArrangeAsync(); + clock.Advance(Window); + Assert.True(kernel.IsDisposing, "nothing was submitted: the idle hub is reclaimed as before"); + + var dispatched = false; + Exception? failure = null; + idle.Track(_ => { dispatched = true; return Observable.Never(); }) + .Subscribe(_ => { }, ex => failure = ex); + + Assert.False(dispatched, "the close and the claim are one ordered decision: a closed hub sets nothing up"); + Assert.IsType(failure); + Assert.Equal(0, idle.InFlight); + } - hub.Received(1).Dispose(); + private async Task<(IMessageHub Kernel, IMessageHub Executor, KernelContainer.IdleState Idle, ManualClock Clock)> ArrangeAsync() + { + var id = Guid.NewGuid().ToString("N"); + var kernel = GetHost().GetHostedHub(new Address("kernelhost", id), c => c)!; + var executor = kernel.GetHostedHub(new Address("kernelExec", id), c => c)!; + await kernel.Started.WaitAsync(TestTimeouts.Convergence, TestContext.Current.CancellationToken); + await executor.Started.WaitAsync(TestTimeouts.Convergence, TestContext.Current.CancellationToken); + var clock = new ManualClock(); + return (kernel, executor, new KernelContainer.IdleState(new WeakReference(kernel), Window, clock), clock); } - [Fact] - public void UnbalancedEnd_NeverCountsBelowZero() + /// + /// A clock that moves only when told to, firing each due one-shot timer at its due instant β€” + /// the injected behind . + /// Single-threaded by construction: the test drives it and every callback runs inline. + /// + private sealed class ManualClock : TimeProvider { - var (state, _, _) = Arrange(); - state.End(); - state.End(); - state.Begin(); + private DateTimeOffset now = DateTimeOffset.UnixEpoch; + private ImmutableList timers = ImmutableList.Empty; + + public override DateTimeOffset GetUtcNow() => now; + + public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period) + { + var timer = new ManualTimer(this, callback, state); + timer.Change(dueTime, period); + timers = timers.Add(timer); + return timer; + } + + public void Advance(TimeSpan by) + { + var target = now + by; + while (timers.Where(t => t.Due <= target).OrderBy(t => t.Due).FirstOrDefault() is { } next) + { + now = next.Due!.Value; + next.Fire(); + } + now = target; + } + } - Assert.Equal(1, state.InFlight); + private sealed class ManualTimer(ManualClock clock, TimerCallback callback, object? state) : ITimer + { + public DateTimeOffset? Due { get; private set; } + + public bool Change(TimeSpan dueTime, TimeSpan period) + { + Due = dueTime == Timeout.InfiniteTimeSpan ? null : clock.GetUtcNow() + dueTime; + return true; + } + + /// One-shot: the due time is consumed BEFORE the callback, which may re-arm it. + public void Fire() + { + Due = null; + callback(state); + } + + public void Dispose() => Due = null; + + public ValueTask DisposeAsync() + { + Dispose(); + return ValueTask.CompletedTask; + } } } From efade15922fe422e83c498d7063d45793a755469 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Tue, 15 Sep 2026 19:31:46 +0200 Subject: [PATCH 3/3] Kernel idle-disconnect: a refused claim does not claim the hub went idle (#4422) Closed is set by the idle tick AND by the hub's own disposal (a recycle, a host teardown), so the refusal names both instead of asserting an idle reclamation it cannot know. Co-Authored-By: Claude Opus 5 (1M context) --- src/MeshWeaver.Kernel.Hub/KernelContainer.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs index b482967656..cfbb01c3ea 100644 --- a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs +++ b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs @@ -274,7 +274,7 @@ public IObservable Track(Func> dispatch) => var claim = new Claim(); if (!ImmutableInterlocked.Update(ref current, s => s.Closed ? s : s with { InFlight = s.InFlight.Add(claim) })) return Observable.Throw(new ObjectDisposedException(nameof(KernelContainer), - "The kernel host was reclaimed after being idle; submit the code again.")); + "The kernel host is shutting down (idle reclamation or disposal); submit the code again.")); return Observable.Defer(() => dispatch(claim)).Finally(() => Release(claim)); });