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();
+ }
+ }
+}