From eb09db7c6958fc7ddf3d526910a643cf47d01c8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Tue, 4 Aug 2026 21:15:54 +0200 Subject: [PATCH] Teardown must quiesce activities before disposing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An activity outlived the mesh that started it. Measured: a 5-second activity, and TeardownAsync returned after 2028ms with the command still running. Teardown already has three phases — DisposalCompleted, IoPoolRegistry.DrainAll(), AsyncDisposeQueue — and an activity falls through all of them: * its trigger returns as soon as the activity exists (by design; it must not block the caller on the work), so the action-block drain has nothing outstanding; * the command runs off-turn via ScheduleOffHubTurn, so it holds no grain turn — holding one would deadlock its own ctx.Log round trips; * SubscribeThroughPool holds a pool permit for the SUBSCRIBE WINDOW ONLY, and deliberately so (it exists to protect the Autofac BeginLifetimeScope). Work that continues past the subscribe — a timer, a continuation, a multi-second command — has already released the permit, so DrainAll() joins nothing. So nothing waited, and work kept executing against a torn-down mesh. When that work touches a collectible ALC's metadata after unload it is a use-after-unload, and the process dies on a bare signal with no managed exception — the shape of the FutuRe.Test exit=139 that reds ~1 CI run in 5. The IoPool's own Dispose comment already names this failure ("the caller may safely unload the node ALCs whose types that work referenced"); activities simply were not covered by it. ActivityTracker counts in-flight runs and teardown waits for zero BEFORE disposing — before, because the run's own Append/Finish writes go back through hubs that must still be alive. Design constraints, all deliberate: * QUIESCE, don't cancel. The run finishes and writes its terminal ActivityLog status. Cancelling would abort work the caller believes is running and leave the activity non-terminal. * Registration releases in Finally — at TERMINAL (succeeded/failed/cancelled), not when the run was merely dispatched. * No locks anywhere. Scan over a scheduler-serialised delta stream: no lock, no Interlocked, not even Observer.Synchronize (which takes one internally). A lock reachable from hub-adjacent code is a deadlock waiting to happen. * The wait belongs to the teardown CALLER, never a hub action block — blocking a turn on this would block the very work it waits for. * Bounded: a run that never settles surfaces as a timeout instead of hanging teardown. * Instance state owned by the mesh, never static — the counter dies with the mesh rather than bleeding into the next test. ActivityOutlivesDisposeTest pins it with a plain delay — no Roslyn, no timing luck: before 2028ms/incomplete, after 5075ms/complete. Co-Authored-By: Claude Opus 5 (1M context) --- src/MeshWeaver.GitSync/ActivityRunner.cs | 9 ++ .../ActivityTracker.cs | 101 ++++++++++++++++++ .../MeshTeardownExtensions.cs | 18 ++++ .../IoPoolServiceCollectionExtensions.cs | 3 + .../ActivityOutlivesDisposeTest.cs | 88 +++++++++++++++ 5 files changed, 219 insertions(+) create mode 100644 src/MeshWeaver.Mesh.Contract/ActivityTracker.cs create mode 100644 test/MeshWeaver.GitSync.Test/ActivityOutlivesDisposeTest.cs diff --git a/src/MeshWeaver.GitSync/ActivityRunner.cs b/src/MeshWeaver.GitSync/ActivityRunner.cs index 0327f0100e..20c73863e0 100644 --- a/src/MeshWeaver.GitSync/ActivityRunner.cs +++ b/src/MeshWeaver.GitSync/ActivityRunner.cs @@ -138,6 +138,13 @@ public static IObservable RunActivity( () => AccessContextScope.FromNode(created, accessService, logger), _ => { + // Register the run as in flight for the WHOLE of its life, released in the + // Finally below — i.e. at TERMINAL (succeeded / failed / cancelled), not when + // it was merely dispatched. Teardown quiesces on this: it stops starting new + // work and waits for what is running, instead of walking away from it while + // the command still executes against a tearing-down mesh. + var runRegistration = hub.ServiceProvider.GetService()?.Track(); + var cts = new CancellationTokenSource(); // Cancel watcher: RequestedStatus = Cancelled trips the command's token // (Activity Control Plane). Subscribed for the life of the run; disposed on @@ -173,6 +180,8 @@ public static IObservable RunActivity( { cancelWatch.Dispose(); cts.Dispose(); + // Terminal: the run is no longer in flight, so teardown may proceed. + runRegistration?.Dispose(); }) .Select(_ => activityPath); })); diff --git a/src/MeshWeaver.Mesh.Contract/ActivityTracker.cs b/src/MeshWeaver.Mesh.Contract/ActivityTracker.cs new file mode 100644 index 0000000000..b96c65bda7 --- /dev/null +++ b/src/MeshWeaver.Mesh.Contract/ActivityTracker.cs @@ -0,0 +1,101 @@ +using System.Reactive; +using System.Reactive.Disposables; +using System.Reactive.Linq; +using System.Reactive.Concurrency; +using System.Reactive.Subjects; + +namespace MeshWeaver.Mesh; + +/// +/// Counts the activity runs currently in flight on this mesh, so teardown can QUIESCE — stop +/// starting new work, let what is running finish, and only then tear down. +/// +/// Why the existing drains do not cover this. Mesh teardown already has three phases +/// (DisposalCompleted → IoPoolRegistry.DrainAll() → AsyncDisposeQueue), but an +/// activity falls through all of them: +/// +/// the trigger returns as soon as the activity exists (by design — it must not +/// block the caller), so the action-block drain has nothing outstanding to wait on; +/// the command runs off-turn via ScheduleOffHubTurn, so it holds no +/// grain turn; +/// SubscribeThroughPool holds its pool permit for the SUBSCRIBE window +/// only — deliberately, to protect the Autofac BeginLifetimeScope. Work that continues +/// past the subscribe (a timer, an await continuation, a multi-second command) has already +/// released the permit, so DrainAll() joins nothing. +/// +/// +/// Measured before this existed: a 5-second activity, and TeardownAsync returned after +/// 2028 ms with the command still running. Work that outlives the mesh keeps executing against a +/// torn-down container — and when it touches a collectible ALC's metadata after unload, the process +/// dies on a bare signal with no managed exception (FutuRe.Test exit=139, ~1 run in 5). +/// +/// Quiesce, do not cancel. The run is allowed to finish and emit its terminal +/// ActivityLog status; teardown waits for that. Cancelling would abort work the caller believes is +/// running and would leave the activity in a non-terminal state. +/// +/// Must not deadlock. The wait belongs to the teardown CALLER, never to a hub action +/// block: a running activity's own Append/Finish writes go back through the hubs, so blocking a hub +/// turn on this would block the very work it waits for. It is also bounded — a run that never +/// settles must surface as a timeout, not hang teardown forever. +/// +/// Instance state owned by the mesh (never static): the counter dies with the mesh +/// rather than bleeding into the next test — see Doc/Architecture/NoStaticState. +/// +public sealed class ActivityTracker : IDisposable +{ + // Deltas in, running count out. Scan accumulates — no lock, no Interlocked, no counter field. + // + // 🚨 No synchronisation primitive anywhere, including Observer.Synchronize: that takes a lock, + // and a lock reachable from hub-adjacent code is a deadlock waiting to happen. Concurrent + // Track()/dispose calls are serialised by QUEUEING them onto a scheduler (ObserveOn), which is + // how Rx serialises without anyone blocking. Scan therefore always runs on one thread and the + // running total cannot interleave. + private readonly Subject deltas = new(); + private readonly EventLoopScheduler scheduler = new(); + private readonly IConnectableObservable counts; + private readonly IDisposable connection; + + /// Initializes a new instance of the class. + public ActivityTracker() + { + // Replay(1) so a late subscriber (teardown) sees the CURRENT count immediately rather than + // waiting for the next change — a teardown on an idle mesh must complete at once. + counts = deltas + .ObserveOn(scheduler) + .Scan(0, (running, delta) => running + delta) + .StartWith(0) + .Replay(1); + connection = counts.Connect(); + } + + /// + /// Live count of in-flight runs, starting with the current value. Emits on every start and + /// every completion, so a consumer waits for zero without polling. + /// + public IObservable InFlightChanges => counts; + + /// + /// Completes once no run is in flight. Emits immediately when the mesh is already idle. + /// + public IObservable WhenIdle => + counts.Where(running => running == 0).Take(1).Select(_ => Unit.Default); + + /// + /// Registers one run as in flight. Dispose the returned handle when the run reaches a TERMINAL + /// state (succeeded or failed) — not when it was merely dispatched. A double dispose does not + /// double-decrement. + /// + public IDisposable Track() + { + deltas.OnNext(1); + return Disposable.Create(() => deltas.OnNext(-1)); + } + + /// + public void Dispose() + { + connection.Dispose(); + deltas.Dispose(); + scheduler.Dispose(); + } +} diff --git a/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs index 35a9fd3d35..81161b6258 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs @@ -49,6 +49,24 @@ public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout) // never resolve DI once disposal has begun (the scope may already be tearing down). var ioPools = mesh.ServiceProvider.GetService(); var asyncDisposeQueue = mesh.ServiceProvider.GetService(); + var activities = mesh.ServiceProvider.GetService(); + + // (0) QUIESCE ACTIVITIES FIRST — before anything is disposed. + // + // An activity falls through all three phases below: its trigger returned as soon as the + // activity existed, its command runs off-turn so it holds no grain turn, and + // SubscribeThroughPool holds a pool permit for the SUBSCRIBE window only — so DrainAll() + // joins nothing once the command continues past it. Measured: a 5s activity, and teardown + // returned after 2028ms with the command still running. + // + // This waits, it does not cancel: the run finishes and writes its terminal ActivityLog + // status. It must run BEFORE Dispose because those Append/Finish writes go back through + // hubs that are still alive. Bounded, so a run that never settles surfaces as a timeout + // rather than hanging teardown forever. + if (activities is not null) + await activities.WhenIdle.Timeout(timeout) + .Catch(_ => Observable.Return(Unit.Default)) + .ToTask(); mesh.Dispose(); diff --git a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs index aaa227ab4c..a79adf784d 100644 --- a/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs @@ -36,6 +36,9 @@ 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(); + // 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(); return services; } } diff --git a/test/MeshWeaver.GitSync.Test/ActivityOutlivesDisposeTest.cs b/test/MeshWeaver.GitSync.Test/ActivityOutlivesDisposeTest.cs new file mode 100644 index 0000000000..f6080dbba7 --- /dev/null +++ b/test/MeshWeaver.GitSync.Test/ActivityOutlivesDisposeTest.cs @@ -0,0 +1,88 @@ +using System; +using System.Reactive; +using System.Reactive.Subjects; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Mesh; +using Xunit; + +namespace MeshWeaver.GitSync.Test; + +/// +/// An activity must never outlive the mesh that started it. +/// +/// The compile that crashed CI on 2026-08-04 (FutuRe.Test exit=139, ~1 run in 5) is one +/// instance of this, but the invariant is general and does not need Roslyn to state: work running +/// inside an activity is work the mesh owns, so disposal must not return while it is still in +/// flight. If it does, the work keeps running against a torn-down mesh — and when that work is +/// emitting or loading a collectible assembly, the ALC unloads underneath it and the process dies +/// on a bare SIGSEGV with no managed exception. +/// +/// Why an activity is the right lifecycle handle: it is a normal hub running inside the +/// message-hub grain, and its ActivityLog reaching a terminal status (Succeeded OR Failed) is the +/// documented signal that it has finished and the hub is free to deactivate +/// (see CompileFinishAndDisposeTest). Nothing else covers it — the activity command deliberately +/// runs OFF the hub turn (ScheduleOffHubTurn), so it holds no grain lock, and a compile +/// additionally leaves the drainable I/O pool for a dedicated thread, so the pool's +/// "teardown cancels + joins in-flight work" guarantee does not reach it either. +/// +/// 🚨 The fix this drives must be CANCEL-THEN-JOIN and must NOT block the action block: the +/// activity's own Append/Finish writes go back through the hub, so a naive blocking join deadlocks +/// against the work it is waiting for — the same class as holding the grain turn across +/// ctx.Log. +/// +public class ActivityOutlivesDisposeTest(ITestOutputHelper output) : GitHubSyncTestBase(output) +{ + /// + /// A deliberately slow activity — a plain delay, no compile, no I/O — started and NOT awaited. + /// Disposal must observe it to terminal before returning. + /// + [Fact(Timeout = 120_000)] + public async Task DisposingTheMesh_WaitsForARunningActivity() + { + var space = "GhDispose" + Guid.NewGuid().ToString("N")[..8]; + await CreateSpace(space, "Dispose-race space"); + + // Reactive signals, not Task primitives: the command entering, and the command finishing. + var entered = new AsyncSubject(); + var completed = 0; + + // Start the activity and DO NOT await it. The trigger returns as soon as the activity + // exists — by design — so this is precisely the window under test. + Mesh.RunActivity(space, ActivityCategory.Import, "Slow probe", + ctx => + { + entered.OnNext(Unit.Default); + entered.OnCompleted(); + return Observable.Timer(TimeSpan.FromSeconds(5)) + .Do(_ => Interlocked.Exchange(ref completed, 1)) + .Select(_ => Unit.Default); + }) + .Subscribe(_ => { }, ex => Output.WriteLine($"activity faulted: {ex.Message}")); + + // Wait for the command to be genuinely running, or the test proves nothing. Reactive + // assertion — FirstAsync() blocks, so it has no place even here. + await entered.Should().Within(TimeSpan.FromSeconds(30)).Emit(); + + // The REAL teardown path. A bare Mesh.Dispose() runs only the synchronous half; the + // documented contract is TeardownAsync — await DisposalCompleted, DrainAll() the pools, + // then quiesce the AsyncDisposeQueue. + var started = DateTimeOffset.UtcNow; + await Mesh.TeardownAsync(TimeSpan.FromSeconds(60)); + var elapsed = DateTimeOffset.UtcNow - started; + Output.WriteLine($"TeardownAsync returned after {elapsed.TotalMilliseconds:F0}ms; " + + $"command completed={Volatile.Read(ref completed) == 1}"); + + // THE CONTRACT: teardown quiesces — it stops new work and waits for what is running to + // finish. It must not return while an activity command is still live: that work keeps + // executing against a torn-down mesh, and when it touches a collectible ALC's metadata + // after unload the process dies on a bare signal with no managed exception. + Volatile.Read(ref completed).Should().Be(1, + $"TeardownAsync returned after {elapsed.TotalMilliseconds:F0}ms while the activity " + + "command was still running. Activity work is mesh-owned: teardown must wait for it, " + + "never walk away from it."); + } +}