diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ShutdownStreamCancellationInterceptor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ShutdownStreamCancellationInterceptor.cs new file mode 100644 index 000000000..a4b27bb2d --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ShutdownStreamCancellationInterceptor.cs @@ -0,0 +1,180 @@ +using Grpc.Core; +using Grpc.Core.Interceptors; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; + +/// +/// Ends every open streaming gRPC call on this host's shared gRPC pipeline when the host starts +/// stopping, and refuses new ones from then on (lmxopcua#906). +/// +/// +/// +/// Why. A server-side 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, i.e. after the host's whole +/// ShutdownTimeout (30 s by default). 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 took 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 is +/// given a context whose token is the call's own token linked with . +/// Handlers that already treat a cancelled call token as "client gone / server shutting down" +/// — 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 +/// mesh telemetry and command streams are long-lived too and would hold the drain the same +/// way. Unary calls are left alone — they are short and finishing them is the point of a drain. +/// +/// +/// 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. +/// +/// +public sealed class ShutdownStreamCancellationInterceptor : Interceptor +{ + 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, falling back to a + // UserState entry for any other — so seed that entry before handing the handler a wrapper. + 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); + } + + /// The key GetHttpContext() falls back to for a context that is not ASP.NET Core's own. + private const string HttpContextKey = "__HttpContext"; + + /// + /// 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/src/Server/ZB.MOM.WW.OtOpcUa.Host/GracefulClusterLeaveService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/GracefulClusterLeaveService.cs new file mode 100644 index 000000000..264b88d3e --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/GracefulClusterLeaveService.cs @@ -0,0 +1,156 @@ +using Akka.Actor; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.OtOpcUa.Host; + +/// +/// Makes a graceful host stop leave the Akka cluster FIRST — before Kestrel drains its open +/// requests — so the peer of a redundant pair takes over the Primary role promptly (lmxopcua#906). +/// +/// +/// +/// Why it is needed. With WebApplication the web server's hosted service starts +/// after every hosted service the app registers, so on stop it runs FIRST, and Akka.Hosting's +/// own StopAsync (which runs ) only runs once Kestrel +/// has finished draining. Kestrel waits for every open request up to the host's +/// ShutdownTimeout (30 s by default) — and the peer's LocalDb sync stream is a +/// long-lived gRPC call that never ends on its own. On QA (2026-11-13) a planned stop of the +/// Primary therefore held the Primary role for ~30 s while the Secondary sat at ServiceLevel 240. +/// +/// +/// How. runs for every lifecycle service +/// before ANY hosted service's StopAsync, Kestrel's included. Here it (1) raises +/// — idempotent; in production +/// RunAsync has already raised it, but a direct IHost.StopAsync raises it only +/// after StoppingAsync — so the long-lived streams are cancelled by +/// and the +/// sees an expected shutdown; then (2) runs +/// and awaits it, bounded by the host's shutdown token. The +/// cluster-leave and singleton hand-over phases therefore complete while Kestrel is still up +/// (so /health/active keeps answering and flips to 503), and only then does Kestrel drain. +/// +/// +/// Why , not Cluster.Leave. A plain +/// leave ends in CoordinatedShutdown with ClusterLeavingReason, which Akka.Hosting's +/// termination hook turns into Environment.ExitCode = -1 — a service supervisor would read +/// a planned stop as a crash. ClrExitReason is exactly the reason Akka.Hosting's own +/// StopAsync uses; its later Run call returns this same, already completed task. +/// +/// +/// Ordering relative to the hosted services that stop AFTER Akka today (the OPC UA endpoint, the +/// LocalDb sync initiator, everything registered before AddAkka) is unchanged: they still +/// stop after the ActorSystem has terminated. Only the web server and the few services registered +/// after AddAkka now stop after it too. +/// +/// +public sealed class GracefulClusterLeaveService : IHostedLifecycleService +{ + private readonly Func _actorSystemAccessor; + private readonly IHostApplicationLifetime _lifetime; + private readonly ILogger _logger; + private volatile bool _started; + + /// Creates the service over a lazy accessor. + /// Lazy accessor; resolved only on stop, never at construction. + /// The host lifetime whose ApplicationStopping this raises. + /// Logger for the leave diagnostics. + public GracefulClusterLeaveService( + Func actorSystemAccessor, + IHostApplicationLifetime lifetime, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(actorSystemAccessor); + ArgumentNullException.ThrowIfNull(lifetime); + ArgumentNullException.ThrowIfNull(logger); + _actorSystemAccessor = actorSystemAccessor; + _lifetime = lifetime; + _logger = logger; + } + + /// + public Task StartingAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task StartedAsync(CancellationToken cancellationToken) + { + // Only a host that got this far has a running ActorSystem worth leaving; a host whose start + // failed earlier must not resolve (and thereby try to build) one on the way down. + _started = true; + return Task.CompletedTask; + } + + /// Raises ApplicationStopping, then runs and awaits CoordinatedShutdown. + /// The host's shutdown token (bounded by ShutdownTimeout). + public async Task StoppingAsync(CancellationToken cancellationToken) + { + _lifetime.StopApplication(); + + if (!_started) + return; + + ActorSystem system; + try + { + system = _actorSystemAccessor(); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Graceful cluster leave skipped: the ActorSystem could not be resolved."); + return; + } + + var started = TimeProvider.System.GetTimestamp(); + _logger.LogInformation( + "Host stopping: leaving the Akka cluster before the web server drains its open requests (lmxopcua#906)."); + try + { + await CoordinatedShutdown.Get(system) + .Run(CoordinatedShutdown.ClrExitReason.Instance) + .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 StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + /// + public Task StoppedAsync(CancellationToken cancellationToken) => Task.CompletedTask; +} + +/// Registration for — shared by Program.cs and its tests. +public static class GracefulClusterLeaveRegistration +{ + /// + /// Registers . Call it after AddAkka; the + /// is resolved lazily, on stop, never at construction. + /// + /// The service collection. + /// The same collection. + public static IServiceCollection AddOtOpcUaGracefulClusterLeave(this IServiceCollection services) + { + ArgumentNullException.ThrowIfNull(services); + services.AddHostedService(sp => new GracefulClusterLeaveService( + () => sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService>())); + return services; + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index b480fa6b2..1a488c873 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -549,6 +549,14 @@ builder.Services.AddAkka("otopcua", (ab, sp) => } }); +// lmxopcua#906: a graceful stop leaves the cluster FIRST. Kestrel's hosted service stops before Akka's +// (WebApplication starts the server last), and Kestrel drains open requests for up to ShutdownTimeout, +// so without this the peer's long-lived LocalDb sync stream held the Primary role for ~30 s on every +// planned stop. Its StoppingAsync runs before ANY hosted service's StopAsync. Registered BEFORE the +// watchdog below only for readability — it raises ApplicationStopping itself, which is what keeps the +// watchdog quiet, so it does not depend on lifecycle-callback order. +builder.Services.AddOtOpcUaGracefulClusterLeave(); + // Simultaneous-cold-start split-brain guard (dark switch Cluster:BootstrapGuard:Enabled, default off). // Registered AFTER AddAkka so its StartAsync runs once Akka's hosted service has built (and, when the // guard is off, already auto-joined) the ActorSystem. When on, the node started with no config seeds @@ -651,6 +659,10 @@ if (hasDriver || hasAdmin) o.Interceptors.Add(); o.Interceptors.Add(); } + // Innermost (after every auth gate): ends open streaming calls — the peer's LocalDb sync stream + // above all — on ApplicationStopping instead of letting Kestrel's drain wait out ShutdownTimeout + // (lmxopcua#906), and refuses new streams once stopping has begun. + o.Interceptors.Add(); }); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/GracefulStopWithOpenSyncStreamTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/GracefulStopWithOpenSyncStreamTests.cs new file mode 100644 index 000000000..d620b0658 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/GracefulStopWithOpenSyncStreamTests.cs @@ -0,0 +1,281 @@ +using System.Diagnostics; +using System.Net; +using Akka.Actor; +using Akka.Cluster; +using Akka.Configuration; +using Akka.Hosting; +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 Shouldly; +using Xunit; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; +using AkkaCluster = Akka.Cluster.Cluster; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb; + +/// +/// lmxopcua#906 — a graceful stop of a 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 passive node is a real — the hosting model whose hosted-service +/// order caused the bug (the web server stops before Akka) — with a real Kestrel h2c listener, the +/// real MapZbLocalDbSync behind the real auth interceptor, a real single-member Akka cluster +/// and the production +/// + . The peer is a real +/// SyncBackgroundService dialling it over loopback. +/// +/// +/// 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 WITHOUT the fix reproduces #906 — 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("LocalDbPairConvergence")] +public sealed class GracefulStopWithOpenSyncStreamTests +{ + private const string ApiKey = "otopcua-906-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); + + static GracefulStopWithOpenSyncStreamTests() => + AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true); + + [Fact] + public async Task GracefulStop_WithOpenSyncStream_LeavesClusterBeforeDrain_AndFinishesPromptly() + { + var result = await RunStopScenarioAsync(withFix: true); + + // Left the cluster, and did so before the web server began draining. + result.MemberRemovedAt.ShouldNotBeNull("the stopping node never left the Akka cluster"); + result.MemberRemovedAt.Value.ShouldBeLessThan(result.ServerStopBeganAt!.Value, + "the node must leave the cluster BEFORE Kestrel drains, so the peer promotes promptly"); + + // Finished well inside the drain budget: the open sync stream did not hold the stop. + result.StopElapsed.ShouldBeLessThan(TimeSpan.FromSeconds(4), + $"stop took {result.StopElapsed.TotalMilliseconds:0} ms against a {ShutdownTimeout.TotalSeconds:0} s drain budget"); + result.MemberRemovedAt.Value.ShouldBeLessThan(TimeSpan.FromSeconds(4)); + } + + [Fact] + public async Task PositiveControl_WithoutTheFix_StopWaitsOutTheDrain_AndLeavesTheClusterLast() + { + var result = await RunStopScenarioAsync(withFix: false); + + // The #906 shape: Kestrel waits for the open stream until the shutdown timeout aborts it, + // and only then does Akka's own hosted service leave the cluster. + result.StopElapsed.ShouldBeGreaterThanOrEqualTo(ShutdownTimeout - TimeSpan.FromSeconds(1)); + result.MemberRemovedAt.ShouldNotBeNull(); + result.MemberRemovedAt.Value.ShouldBeGreaterThan(result.ServerStopBeganAt!.Value); + result.MemberRemovedAt.Value.ShouldBeGreaterThanOrEqualTo(ShutdownTimeout - TimeSpan.FromSeconds(1)); + } + + private sealed record StopResult(TimeSpan StopElapsed, TimeSpan? MemberRemovedAt, TimeSpan? ServerStopBeganAt); + + private static async Task RunStopScenarioAsync(bool withFix) + { + var ct = TestContext.Current.CancellationToken; + var pathA = Path.Combine(Path.GetTempPath(), $"otopcua-906-A-{Guid.NewGuid():N}.db"); + var pathB = Path.Combine(Path.GetTempPath(), $"otopcua-906-B-{Guid.NewGuid():N}.db"); + await using var dbA = BuildDatabaseProvider(pathA); + await using var dbB = BuildDatabaseProvider(pathB); + var clock = new Stopwatch(); + var serverStopBegan = new ServerStopProbe(clock); + IHost? initiator = null; + WebApplication? passive = null; + + try + { + passive = BuildPassiveNode(dbB.GetRequiredService(), serverStopBegan, withFix); + await passive.StartAsync(ct); + + var system = passive.Services.GetRequiredService(); + var cluster = AkkaCluster.Get(system); + cluster.Join(cluster.SelfAddress); + await WaitUntilAsync(() => cluster.State.Members.Any(m => m.Status == MemberStatus.Up), + "the passive 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, ct); + + // The peer's stream is open and holding a session on the passive node. + var passiveStatus = passive.Services.GetRequiredService(); + await WaitUntilAsync(() => passiveStatus.Connected, "the peer's sync stream to be open on the passive node"); + + clock.Start(); + await passive.StopAsync(CancellationToken.None); + var elapsed = clock.Elapsed; + + TestContext.Current.TestOutputHelper?.WriteLine( + $"withFix={withFix}: stop {elapsed.TotalMilliseconds:0} ms; member removed at " + + $"{memberRemovedAt?.TotalMilliseconds:0} ms; web-server drain began at " + + $"{serverStopBegan.BeganAt?.TotalMilliseconds:0} ms (drain budget {ShutdownTimeout.TotalSeconds:0} s)"); + return new StopResult(elapsed, memberRemovedAt, serverStopBegan.BeganAt); + } + 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 BuildPassiveNode(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); + + builder.Services.AddGrpc(o => + { + o.Interceptors.Add(); + if (withFix) + o.Interceptors.Add(); + }); + builder.Services.AddSingleton(db); + builder.Services.AddZbLocalDbReplication(ReplicationConfig(peerAddress: null)); + + builder.Services.AddAkka("otopcua906", (ab, _) => ab.AddHocon( + ConfigurationFactory.ParseString( + "akka.remote.dot-netty.tcp.hostname = \"127.0.0.1\"\n" + + "akka.remote.dot-netty.tcp.public-hostname = \"127.0.0.1\"\n" + + "akka.remote.dot-netty.tcp.port = 0\n" + + "akka.cluster.seed-nodes = []\n" + + "akka.loglevel = \"WARNING\"\n") + .WithFallback(HoconLoader.LoadBaseConfig()), + HoconAddMode.Prepend)); + if (withFix) + builder.Services.AddOtOpcUaGracefulClusterLeave(); + + // 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, CancellationToken ct) + { + var config = ReplicationConfig(peerAddress); + var host = new HostBuilder() + .ConfigureServices(services => + { + services.AddLogging(); + services.AddSingleton(db); + services.AddZbLocalDbReplication(config); + }) + .Build(); + await host.StartAsync(ct); + 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) + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = path, + ["LocalDb:Replication:ApiKey"] = ApiKey, + }) + .Build(); + return new ServiceCollection() + .AddZbLocalDb(config, db => LocalDbSetup.OnReady(db, config)) + .BuildServiceProvider(); + } + + 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, TestContext.Current.CancellationToken); + } + + Assert.Fail($"Timed out waiting for: {because}"); + } + + /// 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(); + } + } +}