Disposal now KNOWS when it finishes: a terminal signal, and drains that cannot lie

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<IoPoolRegistry>();
+            var asyncDisposeQueue = Mesh.ServiceProvider.GetService<AsyncDisposeQueue>();
+            var teardownSignal = Mesh.ServiceProvider.GetService<MeshTeardownSignal>();
+
             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<Unit, Exception>(_ => 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);

src/MeshWeaver.Hosting.Monolith.TestBase/MonolithMeshTestBase.cs

@@ -1283,6 +1283,7 @@ public override async ValueTask DisposeAsync()
             // enqueued. See MeshTeardownExtensions.
             var ioPools = Mesh.ServiceProvider.GetService<IoPoolRegistry>();
             var asyncDisposeQueue = Mesh.ServiceProvider.GetService<AsyncDisposeQueue>();
+            var teardownSignal = Mesh.ServiceProvider.GetService<MeshTeardownSignal>();
             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)

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;

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)
         {

src/MeshWeaver.Import/Implementation/ImportManager.cs

@@ -25,6 +25,21 @@ public class ImportManager
     /// <summary>The message hub this manager serves; source of services and the post target for responses.</summary>
     public IMessageHub Hub { get; }
 
+    /// <summary>
+    /// The mesh-scoped file-system pool the import's async leaves (open + read the data set, run
+    /// the format import) bridge through.
+    ///
+    /// <para>🚨 MUST be the registry pool, never <see cref="IoPool.Unbounded"/> when a mesh
+    /// exists: Unbounded work is INVISIBLE to teardown — <c>CurrentInFlight</c> is hard-wired 0
+    /// and it has no cancellation source, so <c>IoPoolRegistry.DrainAll()</c> 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.</para>
+    /// </summary>
+    private readonly IIoPool ioPool;
+
     /// <summary>
     /// Creates the manager, folding the hub's configured import lambdas into <see cref="Configuration"/>.
     /// </summary>
@@ -37,6 +52,8 @@ public ImportManager(IWorkspace workspace, IMessageHub hub)
 
         Workspace = workspace;
         Hub = hub;
+        ioPool = hub.ServiceProvider.GetService<IoPoolRegistry>()?.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<EntityStore> 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<EntityStore> 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)

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.</item>
 /// </list>
+///
+/// <para>The VERY END of teardown is the <see cref="MeshTeardownSignal"/>: fired here, exactly
+/// once, after every drain phase, carrying the <see cref="TeardownReport"/> 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 <see cref="IMessageHub.DisposalCompleted"/> alone.</para>
 /// </summary>
 public static class MeshTeardownExtensions
 {
@@ -41,7 +47,7 @@ public static class MeshTeardownExtensions
     /// hanging teardown — the underlying bug surfaces elsewhere, e.g.
     /// <c>AnyHubQuiescingTimedOut</c> or a non-zero <see cref="IoPoolRegistry.TotalInFlight"/>).
     /// </summary>
-    public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout)
+    public static async Task<TeardownReport> 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<IoPoolRegistry>();
         var asyncDisposeQueue = mesh.ServiceProvider.GetService<AsyncDisposeQueue>();
         var activities = mesh.ServiceProvider.GetService<ActivityTracker>();
+        var teardownSignal = mesh.ServiceProvider.GetService<MeshTeardownSignal>();
 
         // (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);
     }
 
     /// <summary>
@@ -92,8 +99,16 @@ await activities.WhenIdle.Timeout(timeout)
     ///   (<see cref="AsyncDisposeQueue.DrainAsync"/>), then the caller closes the scope.</item>
     /// </list>
     /// </summary>
-    public static async Task WaitForDisposalAndIoDrainAsync(
-        this IMessageHub mesh, IoPoolRegistry? ioPools, AsyncDisposeQueue? asyncDisposeQueue, TimeSpan timeout)
+    /// <returns>
+    /// The <see cref="TeardownReport"/> — what, if anything, survived the drains. The SAME report
+    /// is fired on <paramref name="teardownSignal"/> (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.
+    /// </returns>
+    public static async Task<TeardownReport> 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;
     }
 }