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 2e191364a6..cfbb01c3ea 100644 --- a/src/MeshWeaver.Kernel.Hub/KernelContainer.cs +++ b/src/MeshWeaver.Kernel.Hub/KernelContainer.cs @@ -41,13 +41,19 @@ public class KernelContainer(IServiceProvider serviceProvider) private readonly object executorHubLock = new(); /// - /// 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-disconnect's shared state β€” set by , read by the + /// submission forwarder. Holds the hub only WEAKLY (see ). /// - private readonly TimeSpan disconnectTimeout = - (serviceProvider.GetService() ?? new KernelHubOptions()).IdleDisconnectTimeout; + private IdleState? idle; + + /// + /// 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 KernelHubOptions options = + serviceProvider.GetService() ?? new KernelHubOptions(); /// /// Hub configuration for the standalone kernel-hub (full mesh-types + routes). @@ -194,23 +200,139 @@ 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), 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); - var idle = disconnectTimeout; + hub.RegisterForDisposal(state); hub.Register(d => { - timer.Change(idle, Timeout.InfiniteTimeSpan); + state.Rearm(); return d; }); } + /// + /// 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 : IDisposable + { + private readonly WeakReference hub; + private readonly TimeSpan window; + private readonly ITimer timer; + private Snapshot current = new(false, ImmutableHashSet.Empty); + + /// 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 current).InFlight.Count; + + /// 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; + + /// + /// 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 is shutting down (idle reclamation or disposal); submit the code again.")); + return Observable.Defer(() => dispatch(claim)).Finally(() => Release(claim)); + }); + + /// A submission answered, failed or was torn down β€” the idle window starts again from now. + private void Release(Claim claim) + { + 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() + { + 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 (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. + /// + internal void Elapsed() + { + if (ImmutableInterlocked.Update(ref current, s => s.Closed || s.InFlight.Any(c => c.Working) ? s : s with { Closed = true })) + { + if (hub.TryGetTarget(out var h)) + h.Dispose(); + return; + } + 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 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) => sp.GetRequiredService>>(); @@ -266,9 +388,14 @@ private IMessageHub GetOrCreateExecutor(IMessageHub publicHub) /// private IMessageDelivery ForwardSubmitCodeRequest(IMessageHub hub, IMessageDelivery request) { - var executor = GetOrCreateExecutor(hub); - - 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) .Subscribe( resp => hub.Post(resp.Message, o => o.ResponseFor(request)), @@ -279,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 bcab445964..0ef683179d 100644 --- a/src/MeshWeaver.Kernel/KernelHubOptions.cs +++ b/src/MeshWeaver.Kernel/KernelHubOptions.cs @@ -26,6 +26,18 @@ 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 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 new file mode 100644 index 0000000000..3b30e8c7c5 --- /dev/null +++ b/test/MeshWeaver.Compiler.Pipeline.Test/KernelIdleDisconnectWhileRunningTest.cs @@ -0,0 +1,208 @@ +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 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. +/// +/// 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(ITestOutputHelper output) : HubTestBase(output) +{ + /// 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 (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 async Task ADeadExecutor_CannotKeepTheHubAliveForever() + { + var (kernel, executor, idle, clock) = await ArrangeAsync(); + idle.Track(claim => { claim.Bind(executor); return Observable.Never(); }).Subscribe(); + + clock.Advance(Window); + Assert.False(kernel.IsDisposing, "kept while the executor is alive"); + + 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 async Task TheWindowElapsingDuringExecutorSetup_KeepsTheHub() + { + 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 async Task ASetupThatThrows_ReleasesItsClaim() + { + var (kernel, _, idle, clock) = await ArrangeAsync(); + Exception? failure = null; + + // 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); + + 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 async Task AClaimAfterTheHubClosed_IsRefused_AndNothingIsSetUp() + { + 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); + } + + 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); + } + + /// + /// 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 + { + 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; + } + } + + 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; + } + } +}