From 57435f32036b46931164aa90fe9cf5328315d52f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Wed, 15 Jul 2026 13:15:54 +0200 Subject: [PATCH 1/3] Fix teardown SIGSEGV: MeshQuery 100% reactive + IIoPool cancel+join drain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause (probe run 29405189603, teardown-stragglers.log): the native `exit=139` in ALC-heavy test hosts is a use-after-unload, NOT the Autofac reflection cache (#467 was orthogonal). MeshQuery ran its provider merge on `SubscribeOn(TaskPoolScheduler.Default)` — unowned thread-pool work that outlives mesh disposal; while `ServiceProvider.Dispose()` unloaded the collectible node ALCs, those stragglers dereferenced freed ALC metadata. The teardown "drain" only WAITED (`IoPoolRegistry.WhenDrained().ToTask()`), never CANCELLED — a live change-feed leaf never completes on its own, so it timed out and the scope disposed under it. Fix (per directive: MeshQuery fully reactive to the IO boundary, IIoPool only for external IO, DisposeAsync ⇒ Dispose): - MeshQuery: delete all three `SubscribeOn(TaskPoolScheduler.Default)`. Every provider already defers external IO to its own `IIoPool` (StorageAdapter) or is pure in-memory `Observable.Return` (StaticNode), so Subscribe never blocks the hub — no unowned straggler; the owned pool is the sole async boundary. - IoPool: link a pool-wide token into every leaf; add `Drain()` = cancel + JOIN (re-acquire all gate permits — no poll/sleep/signal, reuses the gate we own). `Dispose()` now drains synchronously before releasing, so when it returns no pool thread is running and the caller may safely unload node ALCs. - IoPoolRegistry.DrainAll() drains every pool; teardown (MeshTeardownExtensions + MonolithMeshTestBase) calls it synchronously in place of the wait-only `WhenDrained().ToTask()`. - Deterministic test: IoPool.Drain cancels a never-completing in-flight leaf and joins synchronously (CurrentInFlight==0 on return). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../MonolithMeshTestBase.cs | 13 ++-- .../Persistence/Query/MeshQuery.cs | 25 ++++--- .../MeshTeardownExtensions.cs | 19 ++++-- .../Threading/IoPool.cs | 68 +++++++++++++++++-- .../Threading/IoPoolRegistry.cs | 14 ++++ test/MeshWeaver.Hosting.Test/IoPoolTest.cs | 40 +++++++++++ 6 files changed, 149 insertions(+), 30 deletions(-) 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.Hosting/Persistence/Query/MeshQuery.cs b/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs index ea42e519c1..f8dc2c2bef 100644 --- a/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs +++ b/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs @@ -112,13 +112,15 @@ internal static IReadOnlyList MergeQueryNamespaces(ParsedQuery parsed) public IObservable> Query(MeshQueryRequest request) { var matched = SelectMatchingProviders(NamespacesForRequest(request)); + // 100% reactive — NO scheduler offload here. Each provider defers its external I/O to its + // own IIoPool (StorageAdapter) or is pure in-memory (StaticNode), so Subscribe never blocks + // the calling hub's action block. A bare SubscribeOn(TaskPoolScheduler.Default) would push + // the whole merge pipeline onto UNOWNED thread-pool threads whose in-flight work outlives + // mesh disposal and dereferences a collectible node-ALC's freed metadata after unload → the + // native use-after-unload SIGSEGV. The only async boundary is the provider's owned IIoPool. return MergeProviderObservables( - matched.Select(p => (p.Query(request, Options), p.Name)).ToList(), - request) - // Push the upstream subscription onto the thread pool so the - // calling hub's action block isn't blocked while providers open - // their connections / start their change feeds. - .SubscribeOn(System.Reactive.Concurrency.TaskPoolScheduler.Default); + matched.Select(p => (p.Query(request, Options), p.Name)).ToList(), + request); } /// @@ -143,12 +145,10 @@ public IObservable> Query(MeshQueryRequest requ // C return. Same progressive shape as Autocomplete; the brief leading // all-empty frame is the cost of not waiting for the whole fan-out. var streams = matched.Select(p => p.Query(request, Options).StartWith(empty)).ToList(); + // 100% reactive — no scheduler offload (see Query above): the provider's own IIoPool is + // the sole async boundary, so no unowned thread-pool straggler can race a node-ALC unload. return Observable.CombineLatest(streams) - .Select(snapshots => MergeSnapshots(snapshots)) - // Provider Query() opens connection-pool / change-feed - // subscriptions on Subscribe — keep that off the calling hub's - // action block. - .SubscribeOn(System.Reactive.Concurrency.TaskPoolScheduler.Default); + .Select(snapshots => MergeSnapshots(snapshots)); } /// @@ -334,8 +334,7 @@ IObservable> IMeshQueryCore.Query( // Real user: ALWAYS hit the secured provider surface // (validators apply per-result RLS for request.UserId). : p.Query(request, options), p.Name)).ToList(), - request) - .SubscribeOn(System.Reactive.Concurrency.TaskPoolScheduler.Default); + request); } /// 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/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/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() { From 8fa4164c0ef310ba4e849060a3ef84102470a5c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Wed, 15 Jul 2026 18:41:53 +0200 Subject: [PATCH 2/3] Revise: revert MeshQuery offload (regressed AccessContext), drain AsyncDisposeQueue too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Probe run on the first attempt showed two things: 1. Removing MeshQuery's SubscribeOn(TaskPoolScheduler.Default) REGRESSED AccessContext propagation (2× OrleansMarkdownExportTest: "AccessContext must never be null" on the ActivityLog post) and blocked the hub (CancellationAckTest timeout) — that offload captured the ExecutionContext/AccessContext baton at the query-issuer's subscribe. REVERTED. 2. The teardown SIGSEGV is a MULTI-mechanism race, not MeshQuery-specific. With the pool drained, iteration 3 still crashed (CompileLeafStabilityTest); the remaining straggler is AsyncDisposeQueue.DrainAsync — its cleanup ran with CancellationToken.None and DrainAsync TIMED-OUT-AND-ABANDONED in-flight work, which then ran past teardown into the ALC unload. Fix (teardown = cancel + JOIN every async mechanism before ALCs unload): - IoPool.Drain (kept) — cancel + gate-join; DrainAll before ALC unload. - AsyncDisposeQueue: run cleanup under a pool token; DrainAsync now CANCELS on quiesce-timeout then joins, instead of abandoning the in-flight cleanup. - Probe logging fix: NodeAssemblyLoadContext.Dispose wrote the ALC_UNLOAD_PROBE line via the ILogger, which resolves from the disposed container and throws before writing (why the probe never named a culprit). Write to Console.Error directly — the only sink alive at teardown. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Configuration/CompilationCacheService.cs | 6 ++++- .../Persistence/Query/MeshQuery.cs | 25 +++++++++--------- .../Threading/AsyncDisposeQueue.cs | 26 ++++++++++++++----- 3 files changed, 38 insertions(+), 19 deletions(-) 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/Persistence/Query/MeshQuery.cs b/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs index f8dc2c2bef..ea42e519c1 100644 --- a/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs +++ b/src/MeshWeaver.Hosting/Persistence/Query/MeshQuery.cs @@ -112,15 +112,13 @@ internal static IReadOnlyList MergeQueryNamespaces(ParsedQuery parsed) public IObservable> Query(MeshQueryRequest request) { var matched = SelectMatchingProviders(NamespacesForRequest(request)); - // 100% reactive — NO scheduler offload here. Each provider defers its external I/O to its - // own IIoPool (StorageAdapter) or is pure in-memory (StaticNode), so Subscribe never blocks - // the calling hub's action block. A bare SubscribeOn(TaskPoolScheduler.Default) would push - // the whole merge pipeline onto UNOWNED thread-pool threads whose in-flight work outlives - // mesh disposal and dereferences a collectible node-ALC's freed metadata after unload → the - // native use-after-unload SIGSEGV. The only async boundary is the provider's owned IIoPool. return MergeProviderObservables( - matched.Select(p => (p.Query(request, Options), p.Name)).ToList(), - request); + matched.Select(p => (p.Query(request, Options), p.Name)).ToList(), + request) + // Push the upstream subscription onto the thread pool so the + // calling hub's action block isn't blocked while providers open + // their connections / start their change feeds. + .SubscribeOn(System.Reactive.Concurrency.TaskPoolScheduler.Default); } /// @@ -145,10 +143,12 @@ public IObservable> Query(MeshQueryRequest requ // C return. Same progressive shape as Autocomplete; the brief leading // all-empty frame is the cost of not waiting for the whole fan-out. var streams = matched.Select(p => p.Query(request, Options).StartWith(empty)).ToList(); - // 100% reactive — no scheduler offload (see Query above): the provider's own IIoPool is - // the sole async boundary, so no unowned thread-pool straggler can race a node-ALC unload. return Observable.CombineLatest(streams) - .Select(snapshots => MergeSnapshots(snapshots)); + .Select(snapshots => MergeSnapshots(snapshots)) + // Provider Query() opens connection-pool / change-feed + // subscriptions on Subscribe — keep that off the calling hub's + // action block. + .SubscribeOn(System.Reactive.Concurrency.TaskPoolScheduler.Default); } /// @@ -334,7 +334,8 @@ IObservable> IMeshQueryCore.Query( // Real user: ALWAYS hit the secured provider surface // (validators apply per-result RLS for request.UserId). : p.Query(request, options), p.Name)).ToList(), - request); + request) + .SubscribeOn(System.Reactive.Concurrency.TaskPoolScheduler.Default); } /// 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 */ } } } From de906019f8d01bcb0bb4410010225dd4712458d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Wed, 15 Jul 2026 18:44:40 +0200 Subject: [PATCH 3/3] test: AsyncDisposeQueue cancels a wedged cleanup then joins within budget Co-Authored-By: Claude Opus 4.8 (1M context) --- .../AsyncDisposeQueueTest.cs | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) 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() {