diff --git a/src/MeshWeaver.Fixture/HubTestBase.cs b/src/MeshWeaver.Fixture/HubTestBase.cs index 9edd939977..ce9950fa9e 100644 --- a/src/MeshWeaver.Fixture/HubTestBase.cs +++ b/src/MeshWeaver.Fixture/HubTestBase.cs @@ -1,6 +1,8 @@ using System.Reactive; using System.Reactive.Linq; using System.Reactive.Threading.Tasks; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Threading; using MeshWeaver.Messaging; using MeshWeaver.ServiceProvider; using Microsoft.Extensions.DependencyInjection; @@ -155,13 +157,18 @@ public override async ValueTask DisposeAsync() { // Simple timeout - just enough to detect hangs without aggressive intervention var timeoutSeconds = 10; - using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(timeoutSeconds)); // Log which hubs exist before disposal var hostedHubsProperty = Mesh.GetType().GetProperty("HostedHubs"); var hostedHubsValue = hostedHubsProperty?.GetValue(Mesh)?.ToString() ?? "unknown"; Logger.LogInformation("[{DisposalId}] Mesh has {HubCount} hosted hubs", disposalId, hostedHubsValue); + // Capture the mesh-scoped teardown services BEFORE disposal begins — resolving DI + // once the scope is tearing down races its own disposal. + var ioPools = Mesh.ServiceProvider.GetService(); + var asyncDisposeQueue = Mesh.ServiceProvider.GetService(); + var teardownSignal = Mesh.ServiceProvider.GetService(); + if (!Mesh.IsDisposing) { // The fixture owns the mesh, so the fixture must tear it down. This initiation @@ -176,17 +183,23 @@ public override async ValueTask DisposeAsync() } Logger.LogInformation("[{DisposalId}] Mesh is disposing, waiting for completion", disposalId); - // Bridge the reactive completion to a Task once, at this test-teardown edge. - // Catch folds a disposal fault into completion (teardown waits for "done", not why). - // DisposalCompleted replays to late subscribers, so this is race-free. - await Mesh.DisposalCompleted - .Catch(_ => Observable.Return(Unit.Default)) - .FirstOrDefaultAsync() - .ToTask() - .WaitAsync(timeout.Token); - Logger.LogInformation("[{DisposalId}] Mesh disposal completed", disposalId); + // The SAME terminal drain every scenario uses (MeshTeardownExtensions): action blocks + // + round-trips (DisposalCompleted), then cancel+join the offloaded IIoPool leaves, + // then quiesce the AsyncDisposeQueue — and fire the MeshTeardownSignal with the + // report. Waiting on DisposalCompleted ALONE let pooled work outlive the test (the + // straggler class this base previously shared with the pre-fix MonolithMeshTestBase). + var report = await Mesh.WaitForDisposalAndIoDrainAsync( + ioPools, asyncDisposeQueue, TimeSpan.FromSeconds(timeoutSeconds), teardownSignal); + Logger.LogInformation("[{DisposalId}] Mesh disposal completed — {Report}", disposalId, report); + if (!report.Clean) + throw new InvalidOperationException( + $"Teardown left work RUNNING: {report}. A pooled I/O leaf or async cleanup " + + "ignored its cancellation token — fix the leaf; do not widen the drain budget."); } - catch (OperationCanceledException) + // WaitAsync(TimeSpan) inside WaitForDisposalAndIoDrainAsync surfaces a hang as + // TimeoutException (the old inline CancellationTokenSource surfaced it as + // OperationCanceledException) — keep both arms so the hang diagnostics never regress. + catch (Exception timeoutEx) when (timeoutEx is OperationCanceledException or TimeoutException) { Logger.LogError("[{DisposalId}] HANG DETECTED: Mesh disposal timed out after {TimeoutSeconds}s", disposalId, 10); Logger.LogError("[{DisposalId}] Mesh address: {Address}", disposalId, Mesh.Address); diff --git a/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs b/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs index 5b16d54234..888ccff599 100644 --- a/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs +++ b/src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs @@ -1283,6 +1283,7 @@ public override async ValueTask DisposeAsync() // enqueued. See MeshTeardownExtensions. var ioPools = Mesh.ServiceProvider.GetService(); var asyncDisposeQueue = Mesh.ServiceProvider.GetService(); + var teardownSignal = Mesh.ServiceProvider.GetService(); Mesh.Dispose(); TestPhaseTrace(testName, "DISPOSE_INVOKED", sw.ElapsedMilliseconds); @@ -1296,19 +1297,35 @@ public override async ValueTask DisposeAsync() // 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) - { - ioPools.DrainAll(); - if (ioPools.TotalInFlight > 0) - FileOutput.WriteLine( - $"[DISPOSE] {testName}: I/O pools still report {ioPools.TotalInFlight} in-flight after drain"); - } + var leakedIoLeaves = ioPools?.DrainAll() ?? 0; // After all the sync stuff is disposed (and everyone has enqueued their async // cleanup), quiesce the async dispose queue before the scope closes below. - if (asyncDisposeQueue is not null) - await asyncDisposeQueue.DrainAsync(DisposeTimeout); - FileOutput.WriteLine($"[DISPOSE] {testName}: Mesh.Disposal completed in {sw.ElapsedMilliseconds}ms"); - TestPhaseTrace(testName, "DISPOSE_DONE", sw.ElapsedMilliseconds); + var asyncDisposeClean = asyncDisposeQueue is null + || await asyncDisposeQueue.DrainAsync(DisposeTimeout); + + // The terminal signal — DISPOSE_DONE is only true when this report is CLEAN. Fire it + // before the scope disposes below so any subscriber ordering on "all is done" (ALC + // unload hooks, diagnostics) observes the truthful terminal state, dirty or not. + var teardownReport = new TeardownReport(leakedIoLeaves, asyncDisposeClean); + teardownSignal?.SignalCompleted(teardownReport); + FileOutput.WriteLine($"[DISPOSE] {testName}: Mesh.Disposal completed in {sw.ElapsedMilliseconds}ms — {teardownReport}"); + TestPhaseTrace(testName, "DISPOSE_DONE", sw.ElapsedMilliseconds, teardownReport.ToString()); + + // 🚨 A dirty teardown FAILS the class, exactly like the quiesce-leak below. The old + // code logged "I/O pools still report N in-flight" to a per-machine file and carried + // on — and the surviving leaf then dereferenced an unloading node ALC 8 ms into the + // NEXT test's INIT, killing the whole test host with a SIGSEGV that nothing could + // attribute (FutuRe.Test, dump dotnet-3029). Failing HERE names the class that + // leaked, while the evidence still exists. + if (!teardownReport.Clean) + { + TestPhaseTrace(testName, "DISPOSE_DIRTY_TEARDOWN", sw.ElapsedMilliseconds, teardownReport.ToString()); + disposeException = new InvalidOperationException( + $"{testName} teardown left work RUNNING: {teardownReport}. " + + "A pooled I/O leaf or async cleanup ignored its cancellation token; the service " + + "scope (and any collectible node ALC) is about to be torn down over live code — " + + "the use-after-unload SIGSEGV. Fix the leaf that will not cancel; do not widen the drain budget."); + } // Fail the test class' dispose if any hub hit Quiescing timeout. A leaked // Observe subscription that never received its reply is a real bug — @@ -1319,10 +1336,14 @@ public override async ValueTask DisposeAsync() { var summary = Mesh.GetQuiescingTimeoutSummary(); TestPhaseTrace(testName, "DISPOSE_QUIESCE_LEAK", sw.ElapsedMilliseconds, summary); - disposeException = new InvalidOperationException( + var quiesceLeak = new InvalidOperationException( $"{testName} left Observe subscriptions pending past the Quiescing budget. " + $"This is a leaked callback — the test posted a request and never received " + $"(or never awaited) its reply. Pending callbacks at dispose:{Environment.NewLine}{summary}"); + // Never clobber a dirty-teardown failure recorded above — both findings matter. + disposeException = disposeException is null + ? quiesceLeak + : new AggregateException(disposeException, quiesceLeak); } } catch (OperationCanceledException) diff --git a/src/MeshWeaver.Hosting.Orleans/MessageHubGrain.cs b/src/MeshWeaver.Hosting.Orleans/MessageHubGrain.cs index d5846d58fe..658c27dc35 100644 --- a/src/MeshWeaver.Hosting.Orleans/MessageHubGrain.cs +++ b/src/MeshWeaver.Hosting.Orleans/MessageHubGrain.cs @@ -604,6 +604,13 @@ public override async Task OnDeactivateAsync(DeactivationReason reason, Cancella logger.LogError(ex, "Grain {GrainId}: hub disposal failed", grainId); } } + // 🚨 KNOWN GAP (per-ALC accounting): this Unload orders only on the HUB's + // DisposalCompleted above — pooled I/O leaves are mesh-shared, so this grain cannot + // drain them (DrainAll here would cancel every OTHER grain's work). A pooled leaf + // started by this grain's hub that still references this ALC when Unload begins is + // the use-after-unload class. Silo SHUTDOWN is safe (MeshTeardownHostedService drains + // the whole mesh and fires MeshTeardownSignal before the scope dies); a single grain + // deactivating on a LIVE silo needs per-ALC in-flight tracking to close fully. if (loadContext != null) loadContext.Unload(); loadContext = null; diff --git a/src/MeshWeaver.Hosting/MeshTeardownHostedService.cs b/src/MeshWeaver.Hosting/MeshTeardownHostedService.cs index 80b1cdc66c..707347513f 100644 --- a/src/MeshWeaver.Hosting/MeshTeardownHostedService.cs +++ b/src/MeshWeaver.Hosting/MeshTeardownHostedService.cs @@ -51,9 +51,21 @@ public async Task StopAsync(CancellationToken cancellationToken) try { // TeardownAsync captures the mesh-scoped IoPoolRegistry + AsyncDisposeQueue while the - // scope is still alive, disposes the hub, then awaits all three drain phases. - await mesh.TeardownAsync(TeardownTimeout); - logger.LogInformation("MeshTeardownHostedService: mesh {Address} drained cleanly", mesh.Address); + // scope is still alive, disposes the hub, awaits all drain phases, and fires the + // MeshTeardownSignal with the terminal report — the "all is done" notification. + var report = await mesh.TeardownAsync(TeardownTimeout); + if (report.Clean) + logger.LogInformation("MeshTeardownHostedService: mesh {Address} drained cleanly", mesh.Address); + else + // A dirty report means live work survives into scope disposal — the + // use-after-unload class. The host still exits (shutdown must not hang), but + // this is an ERROR: the leaked leaf is a real defect in whatever ignored its + // cancellation token, and this line is the only attribution the crash will get. + logger.LogError( + "MeshTeardownHostedService: mesh {Address} teardown DIRTY — {Report}. " + + "A pooled I/O leaf or async cleanup ignored cancellation and is still running; " + + "the scope is about to dispose over it.", + mesh.Address, report); } catch (Exception ex) { diff --git a/src/MeshWeaver.Import/Implementation/ImportManager.cs b/src/MeshWeaver.Import/Implementation/ImportManager.cs index c3056fa47f..ec62a85002 100644 --- a/src/MeshWeaver.Import/Implementation/ImportManager.cs +++ b/src/MeshWeaver.Import/Implementation/ImportManager.cs @@ -25,6 +25,21 @@ public class ImportManager /// The message hub this manager serves; source of services and the post target for responses. public IMessageHub Hub { get; } + /// + /// The mesh-scoped file-system pool the import's async leaves (open + read the data set, run + /// the format import) bridge through. + /// + /// 🚨 MUST be the registry pool, never when a mesh + /// exists: Unbounded work is INVISIBLE to teardown — CurrentInFlight is hard-wired 0 + /// and it has no cancellation source, so IoPoolRegistry.DrainAll() can neither see nor + /// stop it. An import still running there survives every drain phase, "teardown complete" is + /// signalled anyway, and the leaf then dereferences a collectible node ALC's freed metadata + /// after the scope disposes and unloads it — the FutuRe.Test teardown SIGSEGV (dump + /// dotnet-3029: crash 8 ms into the NEXT test's INIT). The Unbounded fallback is only for a + /// manager constructed outside DI (unit tests without a mesh), where no ALC unload follows. + /// + private readonly IIoPool ioPool; + /// /// Creates the manager, folding the hub's configured import lambdas into . /// @@ -37,6 +52,8 @@ public ImportManager(IWorkspace workspace, IMessageHub hub) Workspace = workspace; Hub = hub; + ioPool = hub.ServiceProvider.GetService()?.Get(IoPoolNames.FileSystem) + ?? IoPool.Unbounded; Configuration = hub.Configuration.GetListOfLambdas().Aggregate(new ImportBuilder(workspace, hub.ServiceProvider), (c, l) => l.Invoke(c)); // Don't initialize the import hub in constructor - do it lazily to avoid timing issues @@ -237,10 +254,11 @@ public IObservable ImportInstances( return ImportWithConfiguredExcelImporter(importRequest, excelConfig, activity); // Read the data set (file/stream I/O leaf), then run the format import — - // composed reactively, each leaf bridged through the IIoPool once. - return IoPool.Unbounded + // composed reactively, each leaf bridged through the mesh-scoped IIoPool once + // (never Unbounded: teardown must be able to see, cancel and join these — see ioPool). + return ioPool .Run(_ => ReadDataSetAsync(importRequest, activity, cancellationToken)) - .SelectMany(read => IoPool.Unbounded + .SelectMany(read => ioPool .Run(_ => read.format.Import(importRequest, read.dataSet, activity, cancellationToken))) .Select(imported => imported!); } @@ -301,7 +319,7 @@ private IObservable ImportWithConfiguredExcelImporter( // bridged through the IIoPool; the Excel parse + EntityStore build // that follow are SYNCHRONOUS (ConfiguredExcelImporter.Import returns // IEnumerable), so they live in a plain Select. No async/await here. ── - return IoPool.Unbounded.Run(_ => streamProvider.Invoke(importRequest)) + return ioPool.Run(_ => streamProvider.Invoke(importRequest)) .Select(stream => { if (stream == null) diff --git a/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs index 81161b6258..149ef853bc 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs @@ -29,6 +29,12 @@ namespace MeshWeaver.Mesh; /// 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. /// +/// +/// The VERY END of teardown is the : fired here, exactly +/// once, after every drain phase, carrying the of what (if +/// anything) survived. Anything that must not run before teardown truly ends — scope disposal, +/// node-ALC unload, the next test's mesh — subscribes to that signal (or uses the report these +/// methods return), never to alone. /// public static class MeshTeardownExtensions { @@ -41,7 +47,7 @@ public static class MeshTeardownExtensions /// hanging teardown — the underlying bug surfaces elsewhere, e.g. /// AnyHubQuiescingTimedOut or a non-zero ). /// - public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout) + public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout) { ArgumentNullException.ThrowIfNull(mesh); @@ -50,6 +56,7 @@ public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout) var ioPools = mesh.ServiceProvider.GetService(); var asyncDisposeQueue = mesh.ServiceProvider.GetService(); var activities = mesh.ServiceProvider.GetService(); + var teardownSignal = mesh.ServiceProvider.GetService(); // (0) QUIESCE ACTIVITIES FIRST — before anything is disposed. // @@ -70,7 +77,7 @@ await activities.WhenIdle.Timeout(timeout) mesh.Dispose(); - await mesh.WaitForDisposalAndIoDrainAsync(ioPools, asyncDisposeQueue, timeout); + return await mesh.WaitForDisposalAndIoDrainAsync(ioPools, asyncDisposeQueue, timeout, teardownSignal); } /// @@ -92,8 +99,16 @@ await activities.WhenIdle.Timeout(timeout) /// (), then the caller closes the scope. /// /// - public static async Task WaitForDisposalAndIoDrainAsync( - this IMessageHub mesh, IoPoolRegistry? ioPools, AsyncDisposeQueue? asyncDisposeQueue, TimeSpan timeout) + /// + /// The — what, if anything, survived the drains. The SAME report + /// is fired on (when one is passed), so out-of-band + /// observers see the identical terminal state the orchestrating caller does. Callers must + /// surface a dirty report (fail the test class, error-log the shutdown) — proceeding + /// silently over live work is the use-after-unload SIGSEGV. + /// + public static async Task WaitForDisposalAndIoDrainAsync( + this IMessageHub mesh, IoPoolRegistry? ioPools, AsyncDisposeQueue? asyncDisposeQueue, + TimeSpan timeout, MeshTeardownSignal? teardownSignal = null) { // (1) Action blocks + message round-trips. await mesh.DisposalCompleted @@ -108,11 +123,19 @@ await mesh.DisposalCompleted // 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(); + var leakedIoLeaves = ioPools?.DrainAll() ?? 0; // (3) After all the sync stuff is disposed (and everyone has enqueued their // async cleanup), quiesce the async dispose queue before the scope closes. - if (asyncDisposeQueue is not null) - await asyncDisposeQueue.DrainAsync(timeout); + var asyncDisposeClean = asyncDisposeQueue is null + || await asyncDisposeQueue.DrainAsync(timeout); + + // (4) The terminal signal — the very end of teardown, all phases accounted. Fired AFTER + // the drains so a subscriber that proceeds on it (scope disposal, ALC unload, next + // test's mesh) never runs concurrently with surviving teardown work — and the report + // tells it when that guarantee could NOT be kept. + var report = new TeardownReport(leakedIoLeaves, asyncDisposeClean); + teardownSignal?.SignalCompleted(report); + return report; } } diff --git a/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs b/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs index b1c869f767..21254e4bec 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/AsyncDisposeQueue.cs @@ -92,18 +92,29 @@ private async Task RunOrphanAsync(Func work) /// 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) + /// + /// true when every posted cleanup ran to completion (possibly cancelled-and-unwound) + /// within the budget — the join is real. false when a cleanup ignored its cancellation + /// token and is STILL RUNNING as this returns: the caller is about to tear down the scope over + /// live work and must surface that, never swallow it. + /// + public async Task DrainAsync(TimeSpan quiesce) { _block.Complete(); try { await _block.Completion.WaitAsync(quiesce).ConfigureAwait(false); - return; // clean quiesce within budget + return true; // clean quiesce within budget } catch (TimeoutException) { /* budget exceeded — cancel + join below */ } _cts.Cancel(); - try { await _block.Completion.WaitAsync(quiesce).ConfigureAwait(false); } + try + { + await _block.Completion.WaitAsync(quiesce).ConfigureAwait(false); + return true; // cancelled cleanups unwound — nothing is still running + } catch (TimeoutException) { /* a cleanup that ignores cancellation is genuinely wedged */ } + return false; } } diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs index 6d2d62c10b..9bda80c924 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPool.cs @@ -227,9 +227,17 @@ public IObservable SubscribeThroughPool(IObservable source) => /// 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() + /// + /// The number of leaves that did NOT unwind within the drain budget — permits that could not be + /// re-acquired because a leaf ignored its cancellation token. 0 means the join is REAL: + /// no pool thread is still running when this returns. Anything else means teardown is about to + /// proceed over live work (the use-after-unload SIGSEGV precondition) — the caller must surface + /// it, never swallow it: a drain that silently gives up is how "disposal completed" becomes a + /// lie and the crash moves 8 ms into the next test's INIT where nothing can attribute it. + /// + public int Drain() { - if (_disposed) return; // gate + cts already gone; nothing in flight to join + if (_disposed) return 0; // gate + cts already gone; nothing in flight to join _poolCts.Cancel(); var acquired = 0; for (var i = 0; i < _maxConcurrency; i++) @@ -239,6 +247,7 @@ public void Drain() } if (acquired > 0) _gate.Release(acquired); + return _maxConcurrency - acquired; } /// diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs index 1bd05f13e9..922579934d 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolRegistry.cs @@ -74,10 +74,17 @@ public IObservable WhenDrained(TimeSpan timeout) => /// 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() + /// + /// The total number of leaves across all pools that did NOT unwind within the drain budget + /// (see ). 0 means the join is real and the caller may proceed + /// to scope disposal / ALC unload. Non-zero means live work survives teardown — surface it. + /// + public int DrainAll() { + var leaked = 0; foreach (var pool in _pools.Values) - pool.Drain(); + leaked += pool.Drain(); + return leaked; } /// Disposes every created pool and clears the registry; called when the mesh is torn down. diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs index a79adf784d..24c37cfc3a 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs @@ -36,6 +36,11 @@ public static IServiceCollection AddIoPools( // The async half of mesh teardown — resources enqueue async cleanup here from // their sync Dispose(); the mesh's DisposeAsync drains it before the scope dies. services.TryAddSingleton(); + // The terminal "all is done" signal — completed exactly once by the teardown + // orchestrator (MeshTeardownExtensions) after EVERY drain phase, carrying the + // TeardownReport. Subscribe to this (never to DisposalCompleted alone) before + // disposing the scope / unloading node ALCs / starting the next mesh. + services.TryAddSingleton(); // Mesh-scoped, so in-flight activity runs are counted per mesh and the counter dies with // it (NoStaticState). Teardown quiesces on this — see ActivityTracker. services.TryAddSingleton(); diff --git a/src/MeshWeaver.Mesh.Contract/Threading/MeshTeardownSignal.cs b/src/MeshWeaver.Mesh.Contract/Threading/MeshTeardownSignal.cs new file mode 100644 index 0000000000..2fa1fad30c --- /dev/null +++ b/src/MeshWeaver.Mesh.Contract/Threading/MeshTeardownSignal.cs @@ -0,0 +1,68 @@ +using System.Reactive.Linq; +using System.Reactive.Subjects; + +namespace MeshWeaver.Mesh.Threading; + +/// +/// What was left behind when mesh teardown reached its terminal point. is the +/// contract: every action block drained, every pooled I/O leaf joined, every async cleanup ran. +/// A non-clean report means live work survived teardown — the precondition of the +/// use-after-unload SIGSEGV (a ThreadPool thread dereferencing a collectible node ALC's freed +/// metadata after the scope disposed and unloaded it). Consumers must SURFACE a dirty report +/// (fail the test class, error-log the host shutdown), never swallow it: a drain that silently +/// gives up is how "disposal completed" becomes a lie and the crash moves a few milliseconds +/// past teardown where nothing can attribute it. +/// +/// Pooled I/O leaves that ignored cancellation and were still +/// running when their drain budget expired (see IoPool.Drain). 0 = the join is real. +/// Whether every cleanup on the +/// ran (or unwound after cancellation) within its quiesce budget. +public sealed record TeardownReport(int LeakedIoLeaves, bool AsyncDisposeClean) +{ + /// True iff nothing survived teardown — the scope may be disposed and node ALCs + /// unloaded with no thread still executing their code. + public bool Clean => LeakedIoLeaves == 0 && AsyncDisposeClean; + + /// One-line summary for logs and failure messages. + public override string ToString() => Clean + ? "teardown clean — all pooled I/O joined, async dispose queue drained" + : $"teardown DIRTY — {LeakedIoLeaves} pooled I/O leaf(s) still running, " + + $"async dispose queue {(AsyncDisposeClean ? "drained" : "still running")}"; +} + +/// +/// The mesh's terminal "all is done" signal — the very END of teardown, strictly after +/// (which only covers the action blocks and +/// message round-trips): it also accounts for the offloaded IIoPool ThreadPool work and the +/// . Completed exactly once by the teardown orchestrator +/// (MeshTeardownExtensions); everything that must not run before teardown truly ends — +/// disposing the service scope, unloading node ALCs, starting the next test's mesh — subscribes +/// here rather than re-deriving "done" from partial signals. +/// +/// Mesh-scoped singleton (never static — dies with the mesh, NoStaticState). Backed by a +/// (1) exactly like MessageHub.disposalCompleted: a +/// subscriber that attaches after teardown already finished still receives the report +/// immediately. Signalling is idempotent via CAS — first report wins. +/// +public sealed class MeshTeardownSignal +{ + private readonly ReplaySubject completed = new(1); + private int signalled; + + /// + /// Fires the final once, then completes — the observable + /// "notification at the very end that all is done". Never errors: a dirty teardown is DATA + /// (the report), surfaced by the consumer, not an Rx fault that would tear subscribers down. + /// + public IObservable Completed => completed.AsObservable(); + + /// Completes exactly once (idempotent CAS) — called only by + /// the teardown orchestrator when every drain phase has finished. + public void SignalCompleted(TeardownReport report) + { + if (Interlocked.CompareExchange(ref signalled, 1, 0) != 0) + return; + completed.OnNext(report); + completed.OnCompleted(); + } +} diff --git a/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs b/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs index a9d7f64838..d02f619e35 100644 --- a/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs +++ b/test/MeshWeaver.Hosting.Test/AsyncDisposeQueueTest.cs @@ -29,11 +29,12 @@ public async Task DrainAsync_runs_every_enqueued_cleanup_and_advances_version_by for (var i = 0; i < items; i++) queue.Enqueue(async _ => { await Task.Yield(); Interlocked.Increment(ref ran); }); - await queue.DrainAsync(Quiesce); + var clean = await queue.DrainAsync(Quiesce); // "putting versions in and checking against own version: old version + N" queue.DrainedVersion.Should().Be(before + items); ran.Should().Be(items); + clean.Should().BeTrue("every cleanup ran within the budget — the join is real"); } // The teardown-SIGSEGV fix: a cleanup that overruns the quiesce budget must be CANCELLED @@ -58,10 +59,33 @@ public async Task DrainAsync_cancels_a_wedged_cleanup_then_joins_within_budget() 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)); + var clean = 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"); + clean.Should().BeTrue("the cancelled cleanup UNWOUND — nothing is still running, so the drain is clean"); + } + + // The report half of the teardown-SIGSEGV fix: a cleanup that IGNORES its cancellation + // token is still running when DrainAsync gives up — and the caller must be able to see + // that, because it is about to dispose the scope (and unload node ALCs) over live code. + // The old void DrainAsync swallowed both timeouts and "teardown complete" became a lie. + [Fact] + public async Task DrainAsync_reports_dirty_when_a_cleanup_ignores_cancellation() + { + var queue = new AsyncDisposeQueue(); + var entered = new TaskCompletionSource(); + var release = new TaskCompletionSource(); + + // A cleanup that ignores ct entirely — the defect class the report exists to name. + queue.Enqueue(async _ => { entered.TrySetResult(); await release.Task; }); + + await entered.Task.WaitAsync(Quiesce, TestContext.Current.CancellationToken); + + var clean = await queue.DrainAsync(TimeSpan.FromMilliseconds(200)); + + clean.Should().BeFalse("the cleanup ignored cancellation and is STILL RUNNING — the caller must surface it"); + release.SetResult(); // let the leaked cleanup finish so it cannot outlive the test } [Fact] diff --git a/test/MeshWeaver.Hosting.Test/MeshTeardownSignalTest.cs b/test/MeshWeaver.Hosting.Test/MeshTeardownSignalTest.cs new file mode 100644 index 0000000000..57bf41d57b --- /dev/null +++ b/test/MeshWeaver.Hosting.Test/MeshTeardownSignalTest.cs @@ -0,0 +1,48 @@ +using System; +using System.Threading.Tasks; +using MeshWeaver.Mesh.Threading; +using MeshWeaver.Reactive.Assertions; +using Xunit; + +namespace MeshWeaver.Hosting.Test; + +/// +/// Pins the terminal "all is done" contract of : fired exactly +/// once by the teardown orchestrator, replayed to late subscribers (so nothing that orders on +/// teardown can miss it), first report wins, and the report's +/// truthfully folds the drain outcomes. +/// +public class MeshTeardownSignalTest +{ + [Fact] + public async Task Completed_replays_the_report_to_a_subscriber_that_attaches_after_teardown() + { + var signal = new MeshTeardownSignal(); + signal.SignalCompleted(new TeardownReport(0, true)); + + // The subscriber attaches AFTER teardown already finished — the exact shape of the + // next test / scope disposal ordering on "all is done". It must still see the report. + var report = await signal.Completed.Should().Within(TimeSpan.FromSeconds(1)).Emit(); + report.Clean.Should().BeTrue(); + } + + [Fact] + public async Task SignalCompleted_is_idempotent_and_the_first_report_wins() + { + var signal = new MeshTeardownSignal(); + signal.SignalCompleted(new TeardownReport(3, false)); + signal.SignalCompleted(new TeardownReport(0, true)); // late duplicate — must not overwrite + + var report = await signal.Completed.Should().Within(TimeSpan.FromSeconds(1)).Emit(); + report.LeakedIoLeaves.Should().Be(3, "the FIRST report is the terminal truth"); + report.Clean.Should().BeFalse(); + } + + [Fact] + public void TeardownReport_is_clean_only_when_nothing_survived() + { + new TeardownReport(0, true).Clean.Should().BeTrue(); + new TeardownReport(1, true).Clean.Should().BeFalse("a leaked I/O leaf is live code surviving teardown"); + new TeardownReport(0, false).Clean.Should().BeFalse("an unfinished async cleanup is live code surviving teardown"); + } +}