diff --git a/src/MeshWeaver.Graph/Configuration/CompilationCacheService.cs b/src/MeshWeaver.Graph/Configuration/CompilationCacheService.cs index 2f9b6af423..222476c23b 100644 --- a/src/MeshWeaver.Graph/Configuration/CompilationCacheService.cs +++ b/src/MeshWeaver.Graph/Configuration/CompilationCacheService.cs @@ -427,7 +427,11 @@ public void Dispose() // an accessor into a collectible node assembly (the exit=139 teardown crash). if (UnloadGcProbe) { - _logger?.LogWarning("ALC_UNLOAD_PROBE forcing GC after unloading {ContextName}", Name); + // Write to Console DIRECTLY, not via _logger: this runs INSIDE ServiceProvider.Dispose() + // where the ILogger (XUnitFileLogger.GetMinLogLevel) resolves from the already-disposed + // container and throws ObjectDisposedException BEFORE writing — so the probe never named + // the culprit. Console is the only sink alive during teardown. + Console.Error.WriteLine($"ALC_UNLOAD_PROBE forcing GC after unloading {Name}"); GC.Collect(); GC.WaitForPendingFinalizers(); GC.Collect(); diff --git a/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs b/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs index dded33973c..21d80ed462 100644 --- a/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs +++ b/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs @@ -1164,15 +1164,18 @@ public override async ValueTask DisposeAsync() await WaitWithProgressAsync(testName, sw, cts.Token); // DisposalCompleted only drains the action blocks + message round-trips. // Offloaded I/O (IIoPool) runs on the ThreadPool independently and is NOT - // covered — wait for it here, BEFORE base.DisposeAsync tears down the - // service scope, so no continuation resolves a disposed Autofac scope - // (the unobserved "catastrophic ObjectDisposedException" class). + // covered — CANCEL + JOIN it here, BEFORE base.DisposeAsync tears down the + // service scope, so no continuation resolves a disposed Autofac scope AND no + // ThreadPool leaf dereferences a collectible node ALC's freed metadata after + // unload (the teardown use-after-unload SIGSEGV). A live change-feed leaf never + // completes on its own, so a WAIT-only drain would time out and let the scope + // dispose under it; DrainAll() cancels the leaf so it stops, then joins. if (ioPools is not null) { - await ioPools.WhenDrained(DisposeTimeout).FirstAsync().ToTask(); + ioPools.DrainAll(); if (ioPools.TotalInFlight > 0) FileOutput.WriteLine( - $"[DISPOSE] {testName}: I/O pools still report {ioPools.TotalInFlight} in-flight after drain wait"); + $"[DISPOSE] {testName}: I/O pools still report {ioPools.TotalInFlight} in-flight after drain"); } // After all the sync stuff is disposed (and everyone has enqueued their async // cleanup), quiesce the async dispose queue before the scope closes below. diff --git a/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs index 94136329af..35a9fd3d35 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs @@ -25,8 +25,9 @@ namespace MeshWeaver.Mesh; /// fires), and /// I/O offloaded onto the ThreadPool through — which /// runs independently of the action block and is NOT tracked by -/// . -/// is the wait for that. +/// . +/// cancels + joins that I/O — a live change-feed leaf never completes on its own, so a +/// wait-without-cancel would time out and let the scope dispose under it. /// /// public static class MeshTeardownExtensions @@ -66,8 +67,8 @@ public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout) /// await — the synchronous/reactive /// disposal: action blocks + message round-trips. Resources enqueue their async /// cleanup onto the during this phase. - /// drain the offloaded ThreadPool I/O the action block doesn't cover - /// (). + /// cancel + join the offloaded ThreadPool I/O the action block doesn't cover + /// (). /// after all the sync stuff is disposed, give the /// a bounded quiesce budget to finish /// (), then the caller closes the scope. @@ -83,9 +84,13 @@ await mesh.DisposalCompleted .ToTask() .WaitAsync(timeout); - // (2) Offloaded ThreadPool I/O — the half DisposalCompleted does not cover. - if (ioPools is not null) - await ioPools.WhenDrained(timeout).FirstAsync().ToTask(); + // (2) Offloaded ThreadPool I/O — the half DisposalCompleted does not cover. CANCEL + JOIN + // synchronously: a live change-feed leaf never completes on its own, so the old + // WhenDrained() (WAIT-only, polled) would time out and let the scope dispose while the + // leaf still runs → its ThreadPool thread dereferences a collectible node ALC's freed + // metadata after unload → native use-after-unload SIGSEGV. DrainAll() cancels every leaf + // so it stops, then joins — no ToTask, no wait-without-cancel. + ioPools?.DrainAll(); // (3) After all the sync stuff is disposed (and everyone has enqueued their // async cleanup), quiesce the async dispose queue before the scope closes. diff --git a/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs b/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs index a52684ea54..b1c869f767 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs @@ -28,6 +28,10 @@ namespace MeshWeaver.Mesh.Threading; public sealed class AsyncDisposeQueue { private readonly ActionBlock> _block; + // Cancels in-flight cleanup when the quiesce budget is exceeded, so a slow cleanup UNWINDS + // (and is then joined) instead of being abandoned mid-flight to run past teardown — where it + // would touch a collectible node ALC's compiled types AFTER the ALC unloads (use-after-unload). + private readonly CancellationTokenSource _cts = new(); private long _drained; /// @@ -42,7 +46,7 @@ public AsyncDisposeQueue() _block = new ActionBlock>( async work => { - try { await work(CancellationToken.None).ConfigureAwait(false); } + try { await work(_cts.Token).ConfigureAwait(false); } catch { /* teardown best-effort — one bad cleanup must not strand the rest */ } finally { Interlocked.Increment(ref _drained); } }, @@ -75,21 +79,31 @@ public void Enqueue(Func asyncCleanup) private async Task RunOrphanAsync(Func work) { - try { await work(CancellationToken.None).ConfigureAwait(false); } + try { await work(_cts.Token).ConfigureAwait(false); } catch { /* best-effort */ } finally { Interlocked.Increment(ref _drained); } } /// /// Terminal drain: stop accepting new items and await every posted item, bounded by - /// . Idempotent. A timeout means a cleanup is genuinely - /// wedged (a separate bug to surface from a stuck resource) — teardown proceeds - /// rather than hanging. + /// . Idempotent. If the budget is exceeded, CANCEL the in-flight + /// cleanup (so it unwinds) then JOIN — no cleanup must still be running (and touching a + /// collectible node ALC's types) when the caller proceeds to unload those ALCs (the teardown + /// use-after-unload SIGSEGV). Only a cleanup that ignores its token can still wedge, and that + /// is a separate bug in that resource; teardown proceeds rather than hanging. /// public async Task DrainAsync(TimeSpan quiesce) { _block.Complete(); + try + { + await _block.Completion.WaitAsync(quiesce).ConfigureAwait(false); + return; // clean quiesce within budget + } + catch (TimeoutException) { /* budget exceeded — cancel + join below */ } + + _cts.Cancel(); try { await _block.Completion.WaitAsync(quiesce).ConfigureAwait(false); } - catch (TimeoutException) { /* wedged cleanup — don't hang teardown */ } + catch (TimeoutException) { /* a cleanup that ignores cancellation is genuinely wedged */ } } } diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs index 06e202015f..19c7b764cf 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs @@ -15,7 +15,16 @@ public sealed class IoPool : IIoPool, IDisposable { private readonly SemaphoreSlim _gate; private readonly TaskFactory _blockingFactory; + private readonly int _maxConcurrency; + // Pool-wide cancellation, linked into every leaf's token. Drain()/Dispose() cancel it so all + // in-flight leaves unwind promptly — the join then knows they will release their gate permits. + private readonly CancellationTokenSource _poolCts = new(); private int _inFlight; + private volatile bool _disposed; + + // Teardown safety net for the drain join: cancellation makes in-flight leaves release in ms, + // so this is effectively never hit — it only bounds a hang if a leaf ignores its token. + private static readonly TimeSpan DrainTimeout = TimeSpan.FromSeconds(30); /// /// Creates a pool whose async gate and sync-blocking scheduler are both capped at @@ -27,6 +36,7 @@ public IoPool(int maxConcurrency) { if (maxConcurrency < 1) throw new ArgumentOutOfRangeException(nameof(maxConcurrency)); + _maxConcurrency = maxConcurrency; _gate = new SemaphoreSlim(maxConcurrency, maxConcurrency); _blockingFactory = new TaskFactory( CancellationToken.None, @@ -50,8 +60,13 @@ public IObservable Invoke(Func> io) // never runs on the calling hub/grain scheduler. (FromAsync's own // scheduler arg only affects notification delivery, not where the // function is invoked — hence the SubscribeOn, matching MeshQuery.) - => Observable.FromAsync(async ct => + => Observable.FromAsync(async subscriberCt => { + // Link the subscriber's token with the pool-wide token so Drain()/Dispose() + // cancels this leaf too — the teardown join relies on every running leaf + // unwinding and releasing its gate permit once the pool is cancelled. + using var linked = CancellationTokenSource.CreateLinkedTokenSource(subscriberCt, _poolCts.Token); + var ct = linked.Token; // WaitAsync(ct) makes acquisition itself cancellable — a dispose // before the slot is granted throws here, before the increment, so // no slot is ever leaked. The ThreadPool thread is released during @@ -76,8 +91,10 @@ public IObservable Invoke(Func> io) /// The cancellable async sequence to enumerate once the gate grants a slot. /// A cold observable that, on subscribe, enumerates the source and emits each element. public IObservable InvokeStream(Func> source) - => Observable.Create(async (observer, ct) => + => Observable.Create(async (observer, subscriberCt) => { + using var linked = CancellationTokenSource.CreateLinkedTokenSource(subscriberCt, _poolCts.Token); + var ct = linked.Token; await _gate.WaitAsync(ct).ConfigureAwait(false); Interlocked.Increment(ref _inFlight); try @@ -103,7 +120,8 @@ public IObservable InvokeStream(Func InvokeBlocking(Func work) => Observable.Create(observer => { - var cts = new CancellationTokenSource(); + // Linked to the pool token so Drain()/Dispose() cancels blocking work too. + var cts = CancellationTokenSource.CreateLinkedTokenSource(_poolCts.Token); _blockingFactory.StartNew(() => { // _inFlight increments only once the scheduler grants a slot — @@ -144,8 +162,48 @@ public IObservable InvokeBlocking(Func work) }); }); - /// Disposes the underlying concurrency gate; called when the owning mesh is torn down. - public void Dispose() => _gate.Dispose(); + /// + /// Cancels every in-flight leaf and JOINS — blocks until they have all unwound — WITHOUT + /// disposing the pool. Called before a collectible node AssemblyLoadContext is unloaded + /// so no pool thread is still executing (or about to dereference) that ALC's compiled types + /// when it is torn down (the teardown use-after-unload SIGSEGV). TERMINAL — it cancels the + /// pool token, so any leaf issued after Drain is cancelled immediately; idempotent (a second + /// call is a safe no-op join). This is a teardown operation; the pool is not reused afterwards. + /// + /// + /// The join uses the gate we already own — no poll, no sleep, no extra signal. A running leaf + /// holds one permit until its finally releases it; once is + /// cancelled every waiting leaf's WaitAsync throws, so no NEW leaf can take a permit. + /// Re-acquiring all permits therefore blocks precisely until the + /// last running leaf has released, then we release them back so the pool stays usable/idempotent. + /// + public void Drain() + { + if (_disposed) return; // gate + cts already gone; nothing in flight to join + _poolCts.Cancel(); + var acquired = 0; + for (var i = 0; i < _maxConcurrency; i++) + { + if (_gate.Wait(DrainTimeout)) acquired++; + else break; // safety net: a leaf ignored its token — don't hang teardown forever + } + if (acquired > 0) + _gate.Release(acquired); + } + + /// + /// Drains in-flight work (see ) then disposes the gate and cancellation + /// source. Synchronous by design: when it returns, no pool thread is running, so the caller may + /// safely unload the node ALCs whose types that work referenced. Called when the mesh is torn down. + /// + public void Dispose() + { + if (_disposed) return; + _disposed = true; + Drain(); + _poolCts.Dispose(); + _gate.Dispose(); + } /// /// A stateless, unbounded fallback pool used when no mesh-scoped pool is diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs index f156bb2d9d..1bd05f13e9 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs @@ -66,6 +66,20 @@ public IObservable WhenDrained(TimeSpan timeout) => .Timeout(timeout) .Catch(_ => Observable.Return(Unit.Default)); + /// + /// Synchronously drains every created pool: cancels all in-flight leaves and JOINS (blocks until + /// they have unwound) — see . Unlike (which + /// only WAITS, so a live change-feed subscription never reaches zero and it times out), this + /// CANCELS the work so it actually stops. Call it between the hub's DisposalCompleted + /// and service-scope disposal so no pooled I/O thread is still executing a collectible node ALC's + /// compiled types when that scope disposes and unloads them (the teardown use-after-unload SIGSEGV). + /// + public void DrainAll() + { + foreach (var pool in _pools.Values) + pool.Drain(); + } + /// Disposes every created pool and clears the registry; called when the mesh is torn down. public void Dispose() { diff --git a/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs b/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs index 589a28499f..a9d7f64838 100644 --- a/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs +++ b/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs @@ -36,6 +36,34 @@ public async Task DrainAsync_runs_every_enqueued_cleanup_and_advances_version_by ran.Should().Be(items); } + // The teardown-SIGSEGV fix: a cleanup that overruns the quiesce budget must be CANCELLED + // (so it unwinds) and JOINED — never abandoned mid-flight to run past teardown, where it + // would touch a collectible node ALC's compiled types after the ALC unloads. + [Fact] + public async Task DrainAsync_cancels_a_wedged_cleanup_then_joins_within_budget() + { + var queue = new AsyncDisposeQueue(); + var before = queue.DrainedVersion; + var cancelled = false; + var entered = new TaskCompletionSource(); + + // A cleanup that never completes on its own (like an in-flight stream / flush). + queue.Enqueue(async ct => + { + entered.TrySetResult(); + try { await Task.Delay(System.Threading.Timeout.Infinite, ct); } + catch (OperationCanceledException) { cancelled = true; throw; } + }); + + await entered.Task.WaitAsync(Quiesce, TestContext.Current.CancellationToken); + + // Budget expires with the cleanup still running → DrainAsync cancels it, then joins. + await queue.DrainAsync(TimeSpan.FromMilliseconds(200)); + + cancelled.Should().BeTrue("DrainAsync cancels a cleanup that overran the quiesce budget"); + queue.DrainedVersion.Should().Be(before + 1, "the cancelled cleanup is JOINED, not abandoned"); + } + [Fact] public async Task Enqueue_from_sync_caller_drains_on_the_background_loop() { diff --git a/test/MeshWeaver.Hosting.Test/IoPoolTest.cs b/test/MeshWeaver.Hosting.Test/IoPoolTest.cs index ec8084d3ca..02e2d09234 100644 --- a/test/MeshWeaver.Hosting.Test/IoPoolTest.cs +++ b/test/MeshWeaver.Hosting.Test/IoPoolTest.cs @@ -151,6 +151,46 @@ public void Invoke_releases_the_slot_when_subscription_disposed_before_completio .Should().BeTrue("disposing the subscription must release the held slot"); } + // The teardown-SIGSEGV fix: Drain must CANCEL every in-flight leaf (a live change-feed + // subscription never completes on its own, so a WAIT-only drain — the old WhenDrained — + // would time out and let the caller unload the node ALCs while the leaf still runs on a + // ThreadPool thread → native use-after-unload) AND JOIN synchronously, so the instant + // Drain returns no pool thread is executing any ALC-compiled code. + [Fact] + public async Task Drain_cancels_in_flight_leaves_and_joins_synchronously() + { + using var pool = new IoPool(2); + using var entered = new ManualResetEventSlim(false); + var cancelled = false; + + pool.Invoke(async ct => + { + entered.Set(); + try { await Task.Delay(System.Threading.Timeout.Infinite, ct); } // never completes on its own + catch (OperationCanceledException) { cancelled = true; throw; } + return 0; + }).Subscribe(_ => { }, _ => { }); + + entered.Wait(Timeout5, TestContext.Current.CancellationToken).Should().BeTrue(); + pool.CurrentInFlight.Should().Be(1); + + pool.Drain(); // cancel the leaf + JOIN — returns only once it has unwound + + pool.CurrentInFlight.Should().Be(0, + "Drain joins synchronously — no spin: when it returns every in-flight leaf has unwound"); + cancelled.Should().BeTrue("Drain cancels in-flight leaves so a never-completing one actually stops"); + + // Drain is TERMINAL (it cancels the pool token) — new work issued after Drain is + // cancelled immediately; there is no in-flight leaf left to reference an unloading ALC. + Func afterDrain = () => + pool.Invoke(_ => Task.FromResult(7)).ToTask(TestContext.Current.CancellationToken); + await afterDrain.Should().ThrowAsync(); + + // Idempotent: a second Drain is a safe no-op join. + pool.Drain(); + pool.CurrentInFlight.Should().Be(0); + } + [Fact] public void Unbounded_fallback_runs_the_leaf_on_the_threadpool() {