From 1a6108c402d50075986eadd85db437ccb52d60f8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 13 Nov 2026 04:16:42 -0500 Subject: [PATCH] fix(host): leave the cluster before Kestrel drains and end open streams on stop (#140) A graceful stop of a site node whose peer holds an open LocalDb sync stream left the Akka cluster ~30 s late (the ScadaBridge half of lmxopcua#906, scadaproj#57). WebApplication stops the web server before AkkaHostedService, whose StopAsync ran CoordinatedShutdown, and Kestrel's graceful drain waited the whole ShutdownTimeout for /localdb_sync.v1.LocalDbSync/Sync: LocalDb 0.3.7 observes only the call token, which Kestrel cancels at the END of the drain. CancelAllStreams only covered SiteStreamGrpcServer's own streams. - AkkaHostedService is now an IHostedLifecycleService. StoppingAsync runs before any hosted service's StopAsync (Kestrel's included): marks the stop intentional, raises ApplicationStopping (so CancelAllStreams still precedes actor teardown), then runs and awaits CoordinatedShutdown(ClrExitReason), bounded by the host shutdown token. StopAsync returns the same completed shutdown. Exit codes unchanged: planned stop exits 0; only a termination outside host shutdown (self-down / #132 shut-out) exits 75. No Akka.Hosting here, so no ClusterLeavingReason/-1 exit-code path is involved. - ShutdownStreamCancellationInterceptor, innermost on both the central and the site AddGrpc pipelines: every streaming call's token is linked with ApplicationStopping; new streams are refused Unavailable while stopping. Covers the LocalDb sync stream until LocalDb 0.3.8 is pinned. Evidence (IntegrationTests/GracefulStopWithOpenSyncStreamTests: real WebApplication + Kestrel h2c + MapZbLocalDbSync behind LocalDbSyncAuthInterceptor + a real LocalDb SyncBackgroundService peer + the production AkkaHostedService forming a single-member cluster, 8 s drain budget, IServer wrapped to timestamp the drain): before 8859 ms stop, member removed at 8834 ms, drain began at 0 ms; after 960 ms stop, member removed at 950 ms, drain began at 959 ms. Mutations (all killed): StoppingAsync skips CoordinatedShutdown; interceptor passes the original context (stop 8003 ms); no Unavailable refusal; StoppingAsync does not raise ApplicationStopping; lifecycle interface dropped; site pipeline loses the interceptor. Release build of the solution: 0 warnings. Host.Tests 733/733, IntegrationTests 133/133, Communication.Tests 843/843. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi --- docs/components/Host.md | 7 + .../Actors/AkkaHostedService.cs | 100 ++++- src/ZB.MOM.WW.ScadaBridge.Host/Program.cs | 17 +- .../ShutdownStreamCancellationInterceptor.cs | 181 +++++++++ ...tdownStreamCancellationInterceptorTests.cs | 189 ++++++++++ .../UnexpectedTerminationTests.cs | 48 +++ .../GracefulStopWithOpenSyncStreamTests.cs | 349 ++++++++++++++++++ 7 files changed, 886 insertions(+), 5 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.Host/ShutdownStreamCancellationInterceptor.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ShutdownStreamCancellationInterceptorTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/GracefulStopWithOpenSyncStreamTests.cs diff --git a/docs/components/Host.md b/docs/components/Host.md index ef85e83b..f6eed4d2 100644 --- a/docs/components/Host.md +++ b/docs/components/Host.md @@ -111,6 +111,13 @@ Before branching on role, `AkkaHostedService.StartAsync` creates one actor uncon Shutdown ordering for the site role is explicit: `IHostApplicationLifetime.ApplicationStopping` fires before `IHostedService.StopAsync`, so `SiteStreamGrpcServer.CancelAllStreams` is called first (clients observe a clean cancellation and reconnect), then `AkkaHostedService` runs `CoordinatedShutdown` and tears down actors. +Since ScadaBridge#140 (the ScadaBridge half of lmxopcua#906) a graceful stop on **both** roles leaves the Akka cluster **before** Kestrel drains. `WebApplication` stops its web server before every hosted service the app registers, and Kestrel's drain waits for open requests up to `HostOptions.ShutdownTimeout` (30 s default) — on a site pair the peer's LocalDb sync stream (`/localdb_sync.v1.LocalDbSync/Sync`) never ends on its own, so the leave used to run ~30 s late and the peer took over that much later. Two pieces fix it: + +- `AkkaHostedService` is an `IHostedLifecycleService`. Its `StoppingAsync` runs before **any** hosted service's `StopAsync` (Kestrel's included): it marks the stop intentional, raises `ApplicationStopping` (idempotent — so `CancelAllStreams` still precedes actor teardown even on a direct `IHost.StopAsync`), then runs and awaits `CoordinatedShutdown` with `ClrExitReason`, bounded by the host's shutdown token. Cluster leave and singleton hand-over complete while Kestrel still serves `/health/active`. `StopAsync` then returns the same completed shutdown. Exit codes are unchanged: a planned stop exits 0; only a termination outside host shutdown (self-down, #132 shut-out restart) exits 75. +- `ShutdownStreamCancellationInterceptor`, registered innermost (after the auth gates) on both the central and site `AddGrpc` pipelines, gives every client-, server- and duplex-streaming call a token linked with `ApplicationStopping` and refuses new streams `Unavailable` once stopping. This ends the LocalDb sync stream (the library only does it itself from LocalDb 0.3.8, scadaproj#57). Unary calls pass through. + +Measured by `IntegrationTests/GracefulStopWithOpenSyncStreamTests` (real `WebApplication` + Kestrel h2c + `MapZbLocalDbSync` + a real LocalDb sync peer + the production `AkkaHostedService`, 8 s drain budget): without the fix the stop took ~8.9 s with the member removed ~8.8 s after the drain began; with it the stop takes ~0.95 s and the member is removed before the drain begins. + ```csharp siteLifetime.ApplicationStopping.Register(() => siteGrpcServer.CancelAllStreams()); ``` diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index a8fe7b8a..998aec43 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -26,7 +26,31 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Actors; /// /// Transport heartbeat is explicitly configured in HOCON from CommunicationOptions. /// -public class AkkaHostedService : IHostedService +/// +/// +/// Leave the cluster before the web server drains (ScadaBridge#140, lmxopcua#906). With +/// WebApplication the web server's hosted service starts after every hosted service the app +/// registers, so on stop it runs FIRST — and Kestrel's graceful drain waits for every open request +/// up to the host's ShutdownTimeout (30 s default). On a site pair the peer's LocalDb sync +/// stream never ends on its own, so CoordinatedShutdown (cluster leave + singleton hand-over), +/// which used to run only in , ran ~30 s late and the peer took over that +/// much later. This service is therefore an : +/// runs before ANY hosted service's StopAsync, Kestrel's +/// included, and runs the shutdown there, while Kestrel is still up (so /health/active +/// keeps answering and flips to 503 as the node leaves). +/// +/// +/// Exit codes are unchanged. The shutdown still runs with +/// — the same reason always +/// used — and marks the stop intentional first, so the +/// watchdog stays quiet and a planned stop exits 0. Only a termination OUTSIDE host shutdown (a +/// self-down or a ScadaBridge#132 shut-out restart) still produces . +/// It deliberately does not call Cluster.Leave: CoordinatedShutdown's own +/// cluster-leave phase performs the leave, and running it through one reason keeps a single +/// shutdown path. +/// +/// +public class AkkaHostedService : IHostedLifecycleService { private readonly IServiceProvider _serviceProvider; private readonly NodeOptions _nodeOptions; @@ -489,7 +513,72 @@ akka {{ /// /// A cancellation token that can be used to cancel the operation. /// A task representing the asynchronous operation. - public async Task StopAsync(CancellationToken cancellationToken) + /// + /// Under a real host, has already run the shutdown and this returns + /// the same, completed CoordinatedShutdown task. Called on its own (tests, hosts that do not + /// honour ) it performs the whole shutdown as before. + /// + public Task StopAsync(CancellationToken cancellationToken) => ShutdownAsync(); + + /// + public Task StartingAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task StartedAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + /// Runs before ANY hosted service's StopAsync — Kestrel's included — and leaves the Akka + /// cluster there (ScadaBridge#140): raises ApplicationStopping (idempotent; in production + /// the service/console lifetime has already raised it, but a direct IHost.StopAsync raises + /// it only after the lifecycle callbacks), then runs and awaits CoordinatedShutdown, bounded by + /// the host's shutdown token. + /// + /// The host's shutdown token (bounded by ShutdownTimeout). + /// A task that completes once the node has left (or the budget ran out). + public async Task StoppingAsync(CancellationToken cancellationToken) + { + // Intentional shutdown first, so the WhenTerminated watchdog stays quiet; then + // ApplicationStopping, so the site's stream cancellation (SiteStreamGrpcServer.CancelAllStreams, + // ShutdownStreamCancellationInterceptor) runs before the actors are torn down — the documented + // order — even when the host is stopped directly rather than through its lifetime. + _stopRequested = true; + _appLifetime?.StopApplication(); + + if (_actorSystem is null) + { + return; + } + + var started = TimeProvider.System.GetTimestamp(); + _logger.LogInformation( + "Host stopping: leaving the Akka cluster before the web server drains its open requests (ScadaBridge#140)."); + try + { + await ShutdownAsync().WaitAsync(cancellationToken).ConfigureAwait(false); + _logger.LogInformation( + "Akka CoordinatedShutdown completed in {ElapsedMs:0} ms; continuing host shutdown.", + TimeProvider.System.GetElapsedTime(started).TotalMilliseconds); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.LogWarning( + "Akka CoordinatedShutdown did not finish within the host shutdown timeout; continuing host shutdown."); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Akka CoordinatedShutdown failed during the graceful leave; continuing host shutdown."); + } + } + + /// + public Task StoppedAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + /// The shutdown proper, shared by and . + /// Idempotent: the tracked-disposable list is drained once and CoordinatedShutdown.Run + /// returns the same task on every call. + /// + private async Task ShutdownAsync() { // Mark this as an intentional shutdown FIRST, so the WhenTerminated // watchdog (wired in GetOrCreateActorSystem) does not mistake the @@ -524,10 +613,13 @@ akka {{ if (_actorSystem != null) { - _logger.LogInformation("Shutting down Akka.NET actor system via CoordinatedShutdown..."); var shutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem); + var first = shutdown.ShutdownReason is null; + if (first) + _logger.LogInformation("Shutting down Akka.NET actor system via CoordinatedShutdown..."); await shutdown.Run(Akka.Actor.CoordinatedShutdown.ClrExitReason.Instance); - _logger.LogInformation("Akka.NET actor system shutdown complete."); + if (first) + _logger.LogInformation("Akka.NET actor system shutdown complete."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs index b6e172ed..8bccaa19 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs @@ -297,6 +297,11 @@ try options.MaxReceiveMessageSize = ZB.MOM.WW.ScadaBridge.Communication.Grpc.GrpcMessageLimits.MaxMessageBytes; options.MaxSendMessageSize = ZB.MOM.WW.ScadaBridge.Communication.Grpc.GrpcMessageLimits.MaxMessageBytes; options.Interceptors.Add(); + // Innermost, after the auth gate (ScadaBridge#140): ends any open streaming call on + // ApplicationStopping instead of letting Kestrel's drain wait out ShutdownTimeout, and + // refuses new streams Unavailable once stopping. Central's control plane is all unary + // today, so this is a backstop for any stream added later. + options.Interceptors.Add(); }); builder.Services.AddSingleton< ZB.MOM.WW.ScadaBridge.Communication.Grpc.CentralControlGrpcService>(); @@ -857,6 +862,12 @@ try options.MaxSendMessageSize = ZB.MOM.WW.ScadaBridge.Communication.Grpc.GrpcMessageLimits.MaxMessageBytes; options.Interceptors.Add(); options.Interceptors.Add(); + // Innermost, after both auth gates (ScadaBridge#140, lmxopcua#906): ends every open + // streaming call — the peer's LocalDb sync stream above all — on ApplicationStopping + // instead of letting Kestrel's drain wait out ShutdownTimeout (30 s), and refuses new + // streams Unavailable once stopping. Host-side until LocalDb 0.3.8 (scadaproj#57) is + // pinned; it stays useful after that as the backstop for every other stream. + options.Interceptors.Add(); }); builder.Services.AddSingleton(); // Site command plane (central→site) — the gRPC peer of SiteStreamGrpcServer. Both share @@ -917,7 +928,11 @@ try // THEN does AkkaHostedService run CoordinatedShutdown and tear down // actors. Without this hand-off, in-flight streams go silent and only // time out via gRPC keepalive (~25 s), violating the documented - // four-step sequence. + // four-step sequence. Since ScadaBridge#140 AkkaHostedService runs that + // CoordinatedShutdown from IHostedLifecycleService.StoppingAsync (raising + // ApplicationStopping first), i.e. BEFORE Kestrel starts draining, and + // ShutdownStreamCancellationInterceptor ends the remaining streams (the + // peer's LocalDb sync stream) so the drain itself is short. var siteLifetime = app.Services.GetRequiredService(); var siteGrpcServer = app.Services.GetRequiredService(); siteLifetime.ApplicationStopping.Register(() => siteGrpcServer.CancelAllStreams()); diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/ShutdownStreamCancellationInterceptor.cs b/src/ZB.MOM.WW.ScadaBridge.Host/ShutdownStreamCancellationInterceptor.cs new file mode 100644 index 00000000..722d70ec --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Host/ShutdownStreamCancellationInterceptor.cs @@ -0,0 +1,181 @@ +using Grpc.Core; +using Grpc.Core.Interceptors; + +namespace ZB.MOM.WW.ScadaBridge.Host; + +/// +/// Ends every open streaming gRPC call on this host when the host starts stopping, and refuses +/// new ones from then on (ScadaBridge#140, the ScadaBridge half of lmxopcua#906). +/// +/// +/// +/// Why. A server-streaming handler observes , +/// which ASP.NET Core cancels only when the client goes away or Kestrel aborts the request — and +/// Kestrel aborts only at the END of its graceful drain, after the host's whole +/// ShutdownTimeout (30 s by default). On a site pair the peer's LocalDb sync stream +/// (/localdb_sync.v1.LocalDbSync/Sync) never ends on its own, so every graceful stop of the +/// node it is connected to held the drain for the full 30 s. Through LocalDb 0.3.7 the library's +/// LocalDbSyncService does not observe ApplicationStopping itself (fixed on scadaproj +/// main 3ddb083, publish tracked as scadaproj#57), so the host supplies it here: the handler +/// gets a context whose token is the call's own token linked with +/// . Handlers that treat a cancelled call +/// token as "client gone / server stopping" — the LocalDb sync service does — then return +/// cleanly, and Kestrel's drain has nothing left to wait for. +/// +/// +/// Scope. Every client-, server- and duplex-streaming call, whatever its service. The site's +/// own SiteStreamService streams are also ended by SiteStreamGrpcServer.CancelAllStreams +/// on ApplicationStopping; this is the generic backstop for every other stream, today and +/// later. Unary calls pass through untouched — they are short, and finishing them is what a drain +/// is for. +/// +/// +/// A stream that arrives once stopping has begun (the peer redialling the moment its stream +/// ended) is refused with — the retryable code — rather than +/// admitted into a node that is about to go away. +/// +/// +/// Exactly one public constructor — Grpc.AspNetCore silently stops invoking an +/// interceptor that has more than one. Register it LAST (innermost), after the auth gates, so an +/// unauthenticated stream is still refused by its gate first. +/// +/// +public sealed class ShutdownStreamCancellationInterceptor : Interceptor +{ + /// The key GetHttpContext() falls back to for a context that is not ASP.NET Core's own. + private const string HttpContextKey = "__HttpContext"; + + private readonly IHostApplicationLifetime _lifetime; + private readonly ILogger _logger; + + /// Creates the interceptor. + /// The host lifetime whose ApplicationStopping ends open streams. + /// Logger for the shutdown diagnostics. + public ShutdownStreamCancellationInterceptor( + IHostApplicationLifetime lifetime, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(lifetime); + ArgumentNullException.ThrowIfNull(logger); + _lifetime = lifetime; + _logger = logger; + } + + /// + public override Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + => RunAsync(context, linked => continuation(requestStream, responseStream, linked)); + + /// + public override Task ServerStreamingServerHandler( + TRequest request, + IServerStreamWriter responseStream, + ServerCallContext context, + ServerStreamingServerMethod continuation) + => RunAsync(context, linked => continuation(request, responseStream, linked)); + + /// + public override async Task ClientStreamingServerHandler( + IAsyncStreamReader requestStream, + ServerCallContext context, + ClientStreamingServerMethod continuation) + { + TResponse response = default!; + await RunAsync(context, async linked => response = await continuation(requestStream, linked).ConfigureAwait(false)) + .ConfigureAwait(false); + return response; + } + + private async Task RunAsync(ServerCallContext context, Func handler) + { + var stopping = _lifetime.ApplicationStopping; + if (stopping.IsCancellationRequested) + { + throw new RpcException(new Status( + StatusCode.Unavailable, "This node is shutting down; retry against its peer or after it restarts.")); + } + + // GetHttpContext() recognises only ASP.NET Core's own context type and otherwise falls back + // to a UserState entry — seed that entry so a handler given the wrapper can still reach it. + try + { + context.UserState.TryAdd(HttpContextKey, context.GetHttpContext()); + } + catch (InvalidOperationException) + { + // Not hosted by ASP.NET Core (a unit test's fake context): nothing to carry over. + } + + using var linked = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, stopping); + using var onStopping = stopping.Register( + static state => + { + var (logger, method) = ((ILogger, string))state!; + logger.LogInformation("Host stopping: ending the open gRPC stream {Method}.", method); + }, + ((ILogger)_logger, context.Method)); + + await handler(new LinkedTokenServerCallContext(context, linked.Token)).ConfigureAwait(false); + } + + /// + /// Delegates everything to the real call context except . + /// is delegated too and carries the __HttpContext + /// entry seeded in RunAsync, so GetHttpContext() still resolves on the wrapper. + /// + internal sealed class LinkedTokenServerCallContext(ServerCallContext inner, CancellationToken token) + : ServerCallContext + { + /// + protected override string MethodCore => inner.Method; + + /// + protected override string HostCore => inner.Host; + + /// + protected override string PeerCore => inner.Peer; + + /// + protected override DateTime DeadlineCore => inner.Deadline; + + /// + protected override Metadata RequestHeadersCore => inner.RequestHeaders; + + /// + protected override CancellationToken CancellationTokenCore => token; + + /// + protected override Metadata ResponseTrailersCore => inner.ResponseTrailers; + + /// + protected override Status StatusCore + { + get => inner.Status; + set => inner.Status = value; + } + + /// + protected override WriteOptions? WriteOptionsCore + { + get => inner.WriteOptions; + set => inner.WriteOptions = value; + } + + /// + protected override AuthContext AuthContextCore => inner.AuthContext; + + /// + protected override IDictionary UserStateCore => inner.UserState; + + /// + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) + => inner.CreatePropagationToken(options); + + /// + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) + => inner.WriteResponseHeadersAsync(responseHeaders); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ShutdownStreamCancellationInterceptorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ShutdownStreamCancellationInterceptorTests.cs new file mode 100644 index 00000000..ee509e53 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ShutdownStreamCancellationInterceptorTests.cs @@ -0,0 +1,189 @@ +using Grpc.Core; +using Microsoft.CodeAnalysis.CSharp; +using Microsoft.CodeAnalysis.CSharp.Syntax; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +/// +/// ScadaBridge#140 — open streaming gRPC calls end on ApplicationStopping instead of holding +/// Kestrel's graceful drain for the whole shutdown timeout, and new streams are refused while stopping. +/// The end-to-end measurement (real Kestrel, real LocalDb sync peer, real Akka node) lives in +/// IntegrationTests/GracefulStopWithOpenSyncStreamTests. +/// +public sealed class ShutdownStreamCancellationInterceptorTests +{ + private static (ShutdownStreamCancellationInterceptor Interceptor, CancellationTokenSource Stopping) Create() + { + var stopping = new CancellationTokenSource(); + var lifetime = Substitute.For(); + lifetime.ApplicationStopping.Returns(stopping.Token); + return (new ShutdownStreamCancellationInterceptor( + lifetime, NullLogger.Instance), stopping); + } + + /// An open server stream's token fires when the host starts stopping, and the call ends. + /// A task that represents the asynchronous operation. + [Fact] + public async Task OpenServerStream_EndsOnApplicationStopping() + { + var (interceptor, stopping) = Create(); + using var callCts = new CancellationTokenSource(); + var context = new FakeServerCallContext("/svc/Stream", callCts.Token); + var entered = new TaskCompletionSource(); + ServerCallContext? seen = null; + + var call = interceptor.ServerStreamingServerHandler( + "req", null!, context, async (_, _, ctx) => + { + seen = ctx; + entered.SetResult(); + await Task.Delay(Timeout.Infinite, ctx.CancellationToken).ContinueWith(_ => { }); + }); + + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.False(call.IsCompleted); + Assert.Equal("/svc/Stream", seen!.Method); + + stopping.Cancel(); + + await call.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.False(callCts.IsCancellationRequested, "the call's own token is not what ended it"); + } + + /// The handler still ends when the client goes away (the call token), as before. + /// A task that represents the asynchronous operation. + [Fact] + public async Task OpenDuplexStream_StillEndsOnTheCallToken() + { + var (interceptor, _) = Create(); + using var callCts = new CancellationTokenSource(); + var context = new FakeServerCallContext("/svc/Duplex", callCts.Token); + + var call = interceptor.DuplexStreamingServerHandler( + null!, null!, context, + (_, _, ctx) => Task.Delay(Timeout.Infinite, ctx.CancellationToken).ContinueWith(_ => { })); + + callCts.Cancel(); + await call.WaitAsync(TimeSpan.FromSeconds(5)); + } + + /// A client stream returns its handler's response through the wrapper. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ClientStream_ReturnsTheHandlerResponse() + { + var (interceptor, _) = Create(); + var context = new FakeServerCallContext("/svc/Upload", CancellationToken.None); + + var response = await interceptor.ClientStreamingServerHandler( + null!, context, (_, _) => Task.FromResult("done")); + + Assert.Equal("done", response); + } + + /// Once stopping has begun, a new stream is refused with the retryable Unavailable. + /// A task that represents the asynchronous operation. + [Fact] + public async Task NewStream_WhileStopping_IsRefusedUnavailable() + { + var (interceptor, stopping) = Create(); + stopping.Cancel(); + var invoked = false; + + var ex = await Assert.ThrowsAsync(() => interceptor.DuplexStreamingServerHandler( + null!, null!, new FakeServerCallContext("/localdb_sync.v1.LocalDbSync/Sync", CancellationToken.None), + (_, _, _) => { invoked = true; return Task.CompletedTask; })); + + Assert.Equal(StatusCode.Unavailable, ex.StatusCode); + Assert.False(invoked); + } + + /// Unary calls are left alone even while stopping — finishing them is what the drain is for. + /// A task that represents the asynchronous operation. + [Fact] + public async Task UnaryCall_WhileStopping_PassesThrough() + { + var (interceptor, stopping) = Create(); + stopping.Cancel(); + + var reply = await interceptor.UnaryServerHandler( + "req", new FakeServerCallContext("/svc/Unary", CancellationToken.None), (_, _) => Task.FromResult("ok")); + + Assert.Equal("ok", reply); + } + + /// + /// Program.cs registers the interceptor LAST (innermost, after the auth gates) on BOTH the + /// central and the site AddGrpc pipelines. + /// + [Fact] + public void Program_RegistersTheInterceptorInnermost_OnBothGrpcPipelines() + { + var program = Path.Combine(FindRepositoryRoot(), "src", "ZB.MOM.WW.ScadaBridge.Host", "Program.cs"); + var root = CSharpSyntaxTree.ParseText(File.ReadAllText(program)).GetRoot(); + + var addGrpcLambdas = root.DescendantNodes().OfType() + .Where(i => i.Expression is MemberAccessExpressionSyntax { Name.Identifier.ValueText: "AddGrpc" }) + .Select(i => i.ArgumentList.Arguments.Single().Expression) + .OfType() + .ToList(); + Assert.Equal(2, addGrpcLambdas.Count); + + foreach (var lambda in addGrpcLambdas) + { + var added = lambda.DescendantNodes().OfType() + .Where(n => n.Identifier.ValueText == "Add" && n.Parent is MemberAccessExpressionSyntax m + && m.Expression.ToString().EndsWith("Interceptors", StringComparison.Ordinal)) + .Select(n => n.TypeArgumentList.Arguments.Single().ToString()) + .ToList(); + Assert.True(added.Count >= 2, $"expected an auth gate and the shutdown interceptor, got [{string.Join(", ", added)}]"); + Assert.Equal(nameof(ShutdownStreamCancellationInterceptor), added[^1]); + } + } + + private static string FindRepositoryRoot() + { + var dir = new DirectoryInfo(AppContext.BaseDirectory); + while (dir is not null && !File.Exists(Path.Combine(dir.FullName, "ZB.MOM.WW.ScadaBridge.slnx"))) + dir = dir.Parent; + return dir?.FullName ?? throw new InvalidOperationException("repository root not found"); + } + + /// Minimal hand-rolled (grpc-dotnet ships no test context). + private sealed class FakeServerCallContext(string method, CancellationToken token) : ServerCallContext + { + /// + protected override string MethodCore => method; + /// + protected override string HostCore => "localhost"; + /// + protected override string PeerCore => "ipv4:127.0.0.1:12345"; + /// + protected override DateTime DeadlineCore => DateTime.UtcNow.AddMinutes(1); + /// + protected override Metadata RequestHeadersCore { get; } = []; + /// + protected override CancellationToken CancellationTokenCore => token; + /// + protected override Metadata ResponseTrailersCore { get; } = []; + /// + protected override Status StatusCore { get; set; } + /// + protected override WriteOptions? WriteOptionsCore { get; set; } + /// + protected override AuthContext AuthContextCore { get; } = + new(null, new Dictionary>()); + /// + protected override IDictionary UserStateCore { get; } = new Dictionary(); + + /// + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) + => throw new NotSupportedException(); + + /// + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/UnexpectedTerminationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/UnexpectedTerminationTests.cs index ebc311c8..bc60eb8c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/UnexpectedTerminationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/UnexpectedTerminationTests.cs @@ -89,6 +89,54 @@ public sealed class UnexpectedTerminationTests Assert.False(service.TerminatedOutsideHostShutdown); } + /// + /// ScadaBridge#140: the lifecycle hook that runs before Kestrel drains raises ApplicationStopping + /// FIRST (so the site's stream cancellation still precedes actor teardown), then completes the + /// CoordinatedShutdown — and a planned stop is never mistaken for a self-down (no exit code 75). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StoppingAsync_RaisesStopping_ThenShutsDown_AndIsNotASelfDown() + { + var lifetime = Substitute.For(); + var service = CreateAkkaHostedService(lifetime); + var system = service.GetOrCreateActorSystem(); + bool? terminatedWhenStoppingRaised = null; + lifetime.When(l => l.StopApplication()).Do(_ => terminatedWhenStoppingRaised = system.WhenTerminated.IsCompleted); + + await service.StoppingAsync(CancellationToken.None); + + Assert.True(system.WhenTerminated.IsCompleted, "StoppingAsync must complete the CoordinatedShutdown"); + Assert.Equal(false, terminatedWhenStoppingRaised); + Assert.Equal(CoordinatedShutdown.ClrExitReason.Instance, CoordinatedShutdown.Get(system).ShutdownReason); + + // The host's later StopAsync is the same, completed shutdown — not a second one. + await service.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(500); + + lifetime.Received(1).StopApplication(); + Assert.False(service.TerminatedOutsideHostShutdown); + } + + /// A host whose Akka service never created its system: StoppingAsync must not create one. + /// A task that represents the asynchronous operation. + [Fact] + public async Task StoppingAsync_WithoutAnActorSystem_DoesNotCreateOne() + { + var lifetime = Substitute.For(); + var service = CreateAkkaHostedService(lifetime); + + await service.StoppingAsync(CancellationToken.None); + + Assert.Null(service.ActorSystem); + lifetime.Received(1).StopApplication(); + } + + /// ScadaBridge#140: the host only calls StoppingAsync on a service that declares the lifecycle interface. + [Fact] + public void AkkaHostedService_IsAHostedLifecycleService() + => Assert.True(typeof(IHostedLifecycleService).IsAssignableFrom(typeof(AkkaHostedService))); + /// /// ScadaBridge#132: the self-down exit code is non-zero (Windows service recovery with /// failureflag 1 restarts only on a non-zero exit) and is only pushed into a Windows service diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/GracefulStopWithOpenSyncStreamTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/GracefulStopWithOpenSyncStreamTests.cs new file mode 100644 index 00000000..d38c8c28 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/GracefulStopWithOpenSyncStreamTests.cs @@ -0,0 +1,349 @@ +using System.Diagnostics; +using System.Net; +using Akka.Actor; +using Akka.Cluster; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Http.Features; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Xunit.Abstractions; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Host; +using ZB.MOM.WW.ScadaBridge.Host.Actors; +using AkkaCluster = Akka.Cluster.Cluster; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests; + +/// +/// ScadaBridge#140 (the ScadaBridge half of lmxopcua#906): a graceful stop of a site node whose +/// peer holds an open LocalDb sync stream must leave the Akka cluster FIRST and finish promptly, +/// not wait out Kestrel's graceful-drain timeout. +/// +/// +/// +/// The stopping node is a real — the hosting model whose hosted-service +/// order caused the bug (the web server stops before the Akka service) — with a real Kestrel h2c +/// listener, the real MapZbLocalDbSync behind the real , +/// the production forming a real single-member Akka cluster, and the +/// production . The peer is a real LocalDb +/// SyncBackgroundService dialling it over loopback and holding its stream open. +/// +/// +/// The web server's is wrapped to timestamp the moment its StopAsync (the +/// drain) begins, so "left the cluster before the drain" is asserted as an ORDER, not inferred from +/// a total duration. +/// +/// +/// The second test is the positive control: the same stack as it was before the fix — the Akka service +/// seen by the host as a plain (so its shutdown runs only in +/// StopAsync) and no stream interceptor — reproduces the bug: the stop waits out the whole +/// shutdown timeout and the cluster is left only after the drain. Without it, a fast stop in the first +/// test could be a harness that never held a stream open at all. +/// +/// +[Collection("LocalDbSitePairConvergence")] +public sealed class GracefulStopWithOpenSyncStreamTests +{ + private const string ApiKey = "scadabridge-140-graceful-stop-key"; + + /// The host's ShutdownTimeout in these tests — Kestrel's drain budget (30 s in production). + private static readonly TimeSpan ShutdownTimeout = TimeSpan.FromSeconds(8); + + private readonly ITestOutputHelper _output; + + static GracefulStopWithOpenSyncStreamTests() => + AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true); + + /// Creates the test class. + /// xunit output, for the measured timings. + public GracefulStopWithOpenSyncStreamTests(ITestOutputHelper output) => _output = output; + + /// With the fix: the node leaves the cluster before the drain begins and stops well inside the budget. + /// A task that represents the asynchronous operation. + [Fact] + public async Task GracefulStop_WithOpenSyncStream_LeavesClusterBeforeDrain_AndFinishesPromptly() + { + var result = await RunStopScenarioAsync(withFix: true); + + // The peer's stream was open and was ended by the stop (not by a peer that never connected). + Assert.True(result.StreamWasOpen, "the peer's sync stream was never open on the stopping node"); + + // Left the cluster, and did so before the web server began draining. + Assert.NotNull(result.MemberRemovedAt); + Assert.NotNull(result.ServerStopBeganAt); + Assert.True(result.MemberRemovedAt < result.ServerStopBeganAt, + $"the node must leave the cluster BEFORE Kestrel drains (removed at {result.MemberRemovedAt?.TotalMilliseconds:0} ms, " + + $"drain began at {result.ServerStopBeganAt?.TotalMilliseconds:0} ms)"); + + // Finished well inside the drain budget: the open sync stream did not hold the stop. + Assert.True(result.StopElapsed < TimeSpan.FromSeconds(4), + $"stop took {result.StopElapsed.TotalMilliseconds:0} ms against a {ShutdownTimeout.TotalSeconds:0} s drain budget"); + + // A planned stop is not a self-down: no exit code 75. + Assert.False(result.TerminatedOutsideHostShutdown); + } + + /// Positive control: the pre-fix wiring reproduces the late leave and the full-budget stop. + /// A task that represents the asynchronous operation. + [Fact] + public async Task PositiveControl_WithoutTheFix_StopWaitsOutTheDrain_AndLeavesTheClusterLast() + { + var result = await RunStopScenarioAsync(withFix: false); + + Assert.True(result.StreamWasOpen); + Assert.True(result.StopElapsed >= ShutdownTimeout - TimeSpan.FromSeconds(1), + $"stop took {result.StopElapsed.TotalMilliseconds:0} ms; expected it to wait out the {ShutdownTimeout.TotalSeconds:0} s drain"); + Assert.NotNull(result.MemberRemovedAt); + Assert.True(result.MemberRemovedAt > result.ServerStopBeganAt); + Assert.True(result.MemberRemovedAt >= ShutdownTimeout - TimeSpan.FromSeconds(1)); + Assert.False(result.TerminatedOutsideHostShutdown); + } + + private sealed record StopResult( + TimeSpan StopElapsed, + TimeSpan? MemberRemovedAt, + TimeSpan? ServerStopBeganAt, + bool StreamWasOpen, + bool TerminatedOutsideHostShutdown); + + private async Task RunStopScenarioAsync(bool withFix) + { + var pathA = Path.Combine(Path.GetTempPath(), $"sb140-A-{Guid.NewGuid():N}.db"); + var pathB = Path.Combine(Path.GetTempPath(), $"sb140-B-{Guid.NewGuid():N}.db"); + await using var dbA = BuildDatabaseProvider(pathA, "node-a"); + await using var dbB = BuildDatabaseProvider(pathB, "node-b"); + var clock = new Stopwatch(); + var probe = new ServerStopProbe(clock); + IHost? initiator = null; + WebApplication? passive = null; + + try + { + passive = BuildStoppingNode(dbB.GetRequiredService(), probe, withFix); + await passive.StartAsync(); + + var akka = passive.Services.GetRequiredService(); + var cluster = AkkaCluster.Get(akka.ActorSystem!); + await WaitUntilAsync(() => cluster.State.Members.Any(m => m.Status == MemberStatus.Up), + "the node to come Up as its own seed"); + + TimeSpan? memberRemovedAt = null; + cluster.RegisterOnMemberRemoved(() => memberRemovedAt ??= clock.Elapsed); + + var address = passive.Services.GetRequiredService() + .Features.Get()!.Addresses.Single(); + initiator = await StartInitiatorAsync(dbA.GetRequiredService(), address); + + // The peer's stream is open and holding a session on the stopping node. + var status = passive.Services.GetRequiredService(); + await WaitUntilAsync(() => status.Connected, "the peer's sync stream to be open on the stopping node"); + var streamWasOpen = status.Connected; + + clock.Start(); + await passive.StopAsync(CancellationToken.None); + var elapsed = clock.Elapsed; + + _output.WriteLine( + $"withFix={withFix}: stop {elapsed.TotalMilliseconds:0} ms; member removed at " + + $"{memberRemovedAt?.TotalMilliseconds:0} ms; web-server drain began at " + + $"{probe.BeganAt?.TotalMilliseconds:0} ms (drain budget {ShutdownTimeout.TotalSeconds:0} s)"); + return new StopResult(elapsed, memberRemovedAt, probe.BeganAt, streamWasOpen, + akka.TerminatedOutsideHostShutdown); + } + finally + { + if (initiator is not null) + { + try { await initiator.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ } + initiator.Dispose(); + } + + if (passive is not null) + await passive.DisposeAsync(); + + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); + foreach (var path in new[] { pathA, pathB }) + foreach (var suffix in new[] { "", "-wal", "-shm" }) + try { File.Delete(path + suffix); } catch { /* best effort */ } + } + } + + private static WebApplication BuildStoppingNode(ILocalDb db, ServerStopProbe probe, bool withFix) + { + var builder = WebApplication.CreateBuilder(); + builder.Environment.ApplicationName = typeof(GracefulStopWithOpenSyncStreamTests).Assembly.GetName().Name!; + builder.Logging.ClearProviders(); + builder.WebHost.ConfigureKestrel(k => + k.Listen(IPAddress.Loopback, 0, o => o.Protocols = HttpProtocols.Http2)); + builder.Services.Configure(o => o.ShutdownTimeout = ShutdownTimeout); + + // The production site interceptor chain: the real auth gate, then (with the fix) the + // stream-cancellation interceptor innermost, exactly as Program.cs orders them. + builder.Services.AddGrpc(o => + { + o.Interceptors.Add(); + if (withFix) + o.Interceptors.Add(); + }); + builder.Services.AddSingleton(db); + builder.Services.AddZbLocalDbReplication(ReplicationConfig(peerAddress: null)); + + // The production AkkaHostedService, registered the way Program/SiteServiceRegistration do. + // Role "StopProbe" (neither Central nor Site) makes StartAsync register only the dead-letter + // monitor; the cluster formation, CoordinatedShutdown and the WhenTerminated watchdog are + // the production code paths. + var port = FreePort(); + builder.Services.AddSingleton(sp => new AkkaHostedService( + sp, + Options.Create(new NodeOptions { Role = "StopProbe", NodeHostname = "127.0.0.1", RemotingPort = port }), + Options.Create(new ClusterOptions + { + SeedNodes = new List { $"akka.tcp://scadabridge@127.0.0.1:{port}" }, + AllowSingleNodeCluster = true, + MinNrOfMembers = 1, + }), + Options.Create(new CommunicationOptions()), + sp.GetRequiredService>(), + sp.GetRequiredService(), + sp.GetRequiredService())); + if (withFix) + builder.Services.AddHostedService(sp => sp.GetRequiredService()); + else + builder.Services.AddHostedService(sp => new PreFixHostedService(sp.GetRequiredService())); + + // Wrap the web server so the start of its drain is timestamped. + var serverDescriptor = builder.Services.Last(d => d.ServiceType == typeof(IServer)); + builder.Services.Remove(serverDescriptor); + builder.Services.AddSingleton(sp => + { + var inner = serverDescriptor.ImplementationFactory is { } factory + ? (IServer)factory(sp) + : (IServer)ActivatorUtilities.CreateInstance(sp, serverDescriptor.ImplementationType!); + return probe.Wrap(inner); + }); + + var app = builder.Build(); + app.MapZbLocalDbSync(); + return app; + } + + private static async Task StartInitiatorAsync(ILocalDb db, string peerAddress) + { + var config = ReplicationConfig(peerAddress); + var host = new HostBuilder() + .ConfigureServices(services => + { + services.AddLogging(); + services.AddSingleton(db); + services.AddZbLocalDbReplication(config); + }) + .Build(); + await host.StartAsync(); + return host; + } + + private static IConfiguration ReplicationConfig(string? peerAddress) + { + var values = new Dictionary + { + ["LocalDb:Replication:FlushInterval"] = "00:00:00.050", + ["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02", + ["LocalDb:Replication:ApiKey"] = ApiKey, + }; + if (peerAddress is not null) + values["LocalDb:Replication:PeerAddress"] = peerAddress; + return new ConfigurationBuilder().AddInMemoryCollection(values).Build(); + } + + private static ServiceProvider BuildDatabaseProvider(string path, string nodeName) + { + string Absent() => Path.Combine(Path.GetTempPath(), $"absent-{Guid.NewGuid():N}.db"); + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = path, + ["ScadaBridge:Node:NodeName"] = nodeName, + ["LocalDb:Replication:ApiKey"] = ApiKey, + // Legacy migrators pointed at paths that do not exist, so they no-op. + ["ScadaBridge:OperationTracking:ConnectionString"] = $"Data Source={Absent()}", + ["ScadaBridge:SiteEventLog:DatabasePath"] = Absent(), + ["ScadaBridge:StoreAndForward:SqliteDbPath"] = Absent(), + ["ScadaBridge:Database:SiteDbPath"] = Absent(), + }) + .Build(); + + return new ServiceCollection() + .AddZbLocalDb(config, db => SiteLocalDbSetup.OnReady(db, config)) + .BuildServiceProvider(); + } + + private static int FreePort() + { + var l = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); + l.Start(); + var p = ((IPEndPoint)l.LocalEndpoint).Port; + l.Stop(); + return p; + } + + private static async Task WaitUntilAsync(Func condition, string because) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); + while (DateTime.UtcNow < deadline) + { + if (condition()) + return; + await Task.Delay(50); + } + + Assert.Fail($"Timed out waiting for: {because}"); + } + + /// + /// The pre-fix registration shape: the host sees the Akka service only as an + /// , so its shutdown runs in StopAsync, after Kestrel's. + /// + private sealed class PreFixHostedService(AkkaHostedService inner) : IHostedService + { + public Task StartAsync(CancellationToken cancellationToken) => inner.StartAsync(cancellationToken); + + public Task StopAsync(CancellationToken cancellationToken) => inner.StopAsync(cancellationToken); + } + + /// Records when the wrapped web server's StopAsync (its graceful drain) begins. + private sealed class ServerStopProbe(Stopwatch clock) + { + public TimeSpan? BeganAt { get; private set; } + + private void MarkBegan() => BeganAt ??= clock.Elapsed; + + public IServer Wrap(IServer inner) => new RecordingServer(inner, this); + + private sealed class RecordingServer(IServer inner, ServerStopProbe probe) : IServer + { + public IFeatureCollection Features => inner.Features; + + public Task StartAsync(IHttpApplication application, CancellationToken cancellationToken) + where TContext : notnull + => inner.StartAsync(application, cancellationToken); + + public Task StopAsync(CancellationToken cancellationToken) + { + probe.MarkBegan(); + return inner.StopAsync(cancellationToken); + } + + public void Dispose() => inner.Dispose(); + } + } +}