fix(localdb): adapter review fixes — deterministic pump teardown, always-hosted initiator, backoff clamp

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
2026-07-17 23:13:02 -04:00
parent a64fc26391
commit e92cdc6d3a
6 changed files with 148 additions and 28 deletions
@@ -14,10 +14,10 @@ public static class ReplicationServiceCollectionExtensions
    /// Binds and validates <see cref="ReplicationOptions"/> from the <c>"LocalDb:Replication"</c>
    /// configuration section, registers the passive <see cref="LocalDbSyncService"/> as a singleton
    /// (its single-stream guard must span calls), and registers the initiating
-    /// <see cref="SyncBackgroundService"/> as a hosted service ONLY when a peer address is configured
-    /// (a passive node leaves it empty and never dials). Idempotent. Requires
-    /// <see cref="LocalDbServiceCollectionExtensions.AddZbLocalDb"/> to have been called for the
-    /// <see cref="ILocalDb"/> the adapters depend on.
+    /// <see cref="SyncBackgroundService"/> as a hosted service — on a passive node (no
+    /// <see cref="ReplicationOptions.PeerAddress"/>) it starts and immediately idles. Idempotent.
+    /// Requires <see cref="LocalDbServiceCollectionExtensions.AddZbLocalDb"/> to have been called
+    /// for the <see cref="ILocalDb"/> the adapters depend on.
    /// </summary>
    public static IServiceCollection AddZbLocalDbReplication(
        this IServiceCollection services, IConfiguration configuration)
@@ -33,10 +33,6 @@ public static class ReplicationServiceCollectionExtensions
            ServiceDescriptor.Singleton<IValidateOptions<ReplicationOptions>, ReplicationOptionsValidator>());

        services.AddSingleton<LocalDbSyncService>();
-
-        // Read the section directly (before options are built) to decide whether this node initiates.
-        var peerAddress = configuration.GetSection("LocalDb:Replication")["PeerAddress"];
-        if (!string.IsNullOrEmpty(peerAddress))
        services.AddHostedService<SyncBackgroundService>();

        return services;
@@ -10,6 +10,16 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal;
 /// </summary>
 internal static class SyncSessionFactory
 {
+    /// <summary>
+    /// The engine binds to <see cref="SqliteLocalDb"/> internals (NodeId, Clock, replicated-table
+    /// digests), so both adapters require the concrete type AddZbLocalDb registers behind ILocalDb.
+    /// </summary>
+    public static SqliteLocalDb RequireSqlite(ILocalDb db, string consumer) =>
+        db as SqliteLocalDb
+        ?? throw new InvalidOperationException(
+            $"{consumer} requires the ILocalDb registered by AddZbLocalDb (a {nameof(SqliteLocalDb)}); " +
+            $"got {db.GetType().FullName}.");
+
    public static SyncSession Create(SqliteLocalDb db, ReplicationOptions options, ILoggerFactory loggerFactory)
    {
        var store = new OplogStore(db, options);
@@ -17,18 +17,17 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
    private readonly SqliteLocalDb _db;
    private readonly ReplicationOptions _options;
    private readonly ILoggerFactory _loggerFactory;
+    private readonly ILogger<LocalDbSyncService> _logger;

    // 0 = idle, 1 = a stream is active. Interlocked-guarded across concurrent calls (singleton service).
    private int _active;

    public LocalDbSyncService(ILocalDb db, IOptions<ReplicationOptions> options, ILoggerFactory loggerFactory)
    {
-        _db = db as SqliteLocalDb
-            ?? throw new InvalidOperationException(
-                $"LocalDbSyncService requires the ILocalDb registered by AddZbLocalDb (a {nameof(SqliteLocalDb)}); " +
-                $"got {db.GetType().FullName}.");
+        _db = SyncSessionFactory.RequireSqlite(db, nameof(LocalDbSyncService));
        _options = options.Value;
        _loggerFactory = loggerFactory;
+        _logger = loggerFactory.CreateLogger<LocalDbSyncService>();
    }

    public override async Task Sync(
@@ -41,11 +40,30 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase

        try
        {
-            var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, context.CancellationToken);
+            // The linked source exists so the reader pump is deterministically torn down on EVERY
+            // exit path — including a session fault for a local (non-transport) reason, where the
+            // call token never fires and the pump would otherwise keep reading a stream whose
+            // handler has exited.
+            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
+            var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, linkedCts.Token);
            var session = SyncSessionFactory.Create(_db, _options, _loggerFactory);
-            await session.RunAsync(duplex, context.CancellationToken);
+            try
+            {
+                await session.RunAsync(duplex, linkedCts.Token);
+            }
+            finally
+            {
+                linkedCts.Cancel();
+                try
+                {
                    await readerTask;
                }
+                catch (Exception ex)
+                {
+                    _logger.LogDebug(ex, "Sync reader pump ended during teardown.");
+                }
+            }
+        }
        catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested)
        {
            // Client disconnect / server shutdown: the normal end of a long-lived streaming RPC.
@@ -9,6 +9,9 @@ namespace ZB.MOM.WW.LocalDb.Replication;
 /// </summary>
 public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOptions>
 {
+    // Also what keeps SyncBackgroundService's unchecked backoff doubling overflow-safe.
+    private static readonly TimeSpan ReconnectBackoffCeiling = TimeSpan.FromDays(1);
+
    public ValidateOptionsResult Validate(string? name, ReplicationOptions options)
    {
        var failures = new List<string>();
@@ -31,6 +34,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOp
            failures.Add($"LocalDb:Replication:TombstoneRetention must be greater than zero; got {options.TombstoneRetention}.");
        if (options.ReconnectBackoffMax <= TimeSpan.Zero)
            failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be greater than zero; got {options.ReconnectBackoffMax}.");
+        else if (options.ReconnectBackoffMax > ReconnectBackoffCeiling)
+            failures.Add($"LocalDb:Replication:ReconnectBackoffMax must be at most {ReconnectBackoffCeiling}; got {options.ReconnectBackoffMax}.");
        if (options.MaxHlcDriftAhead <= TimeSpan.Zero)
            failures.Add($"LocalDb:Replication:MaxHlcDriftAhead must be greater than zero; got {options.MaxHlcDriftAhead}.");

@@ -12,7 +12,8 @@ namespace ZB.MOM.WW.LocalDb.Replication;
 /// <summary>
 /// The initiating (client) side of replication: a hosted service that dials the configured peer,
 /// runs the symmetric <see cref="SyncSession"/> over the outbound bidirectional stream, and
-/// reconnects with jittered exponential backoff whenever the session ends or faults.
+/// reconnects with jittered exponential backoff whenever the session ends or faults. Always
+/// registered; on a passive node (no <see cref="ReplicationOptions.PeerAddress"/>) it idles.
 /// </summary>
 public sealed class SyncBackgroundService : BackgroundService
 {
@@ -34,10 +35,7 @@ public sealed class SyncBackgroundService : BackgroundService

    public SyncBackgroundService(ILocalDb db, IOptions<ReplicationOptions> options, ILoggerFactory loggerFactory)
    {
-        _db = db as SqliteLocalDb
-            ?? throw new InvalidOperationException(
-                $"SyncBackgroundService requires the ILocalDb registered by AddZbLocalDb (a {nameof(SqliteLocalDb)}); " +
-                $"got {db.GetType().FullName}.");
+        _db = SyncSessionFactory.RequireSqlite(db, nameof(SyncBackgroundService));
        _options = options.Value;
        _loggerFactory = loggerFactory;
        _logger = loggerFactory.CreateLogger<SyncBackgroundService>();
@@ -45,6 +43,12 @@ public sealed class SyncBackgroundService : BackgroundService

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
+        if (string.IsNullOrEmpty(_options.PeerAddress))
+        {
+            _logger.LogInformation("passive node — no peer configured, sync initiator idle");
+            return;
+        }
+
        var backoff = InitialBackoff;
        while (!stoppingToken.IsCancellationRequested)
        {
@@ -65,13 +69,31 @@ public sealed class SyncBackgroundService : BackgroundService
                    headers.Add("Authorization", $"Bearer {_options.ApiKey}");

                using var call = client.Sync(headers, cancellationToken: stoppingToken);
-                var (duplex, readerTask) = GrpcSyncDuplex.Create(call.ResponseStream, call.RequestStream, stoppingToken);
+                // The linked source exists so the reader pump is deterministically torn down on
+                // EVERY exit path — including a session fault for a local (non-transport) reason,
+                // where the stopping token never fires and the pump would otherwise keep reading a
+                // stream this iteration has abandoned.
+                using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
+                var (duplex, readerTask) = GrpcSyncDuplex.Create(call.ResponseStream, call.RequestStream, linkedCts.Token);
                var session = SyncSessionFactory.Create(_db, _options, _loggerFactory);
-
-                await session.RunAsync(duplex, stoppingToken);
+                try
+                {
+                    await session.RunAsync(duplex, linkedCts.Token);
                    await call.RequestStream.CompleteAsync();
+                }
+                finally
+                {
+                    linkedCts.Cancel();
+                    try
+                    {
                        await readerTask;
                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogDebug(ex, "Sync reader pump ended during teardown.");
+                    }
+                }
+            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
@@ -107,6 +129,8 @@ public sealed class SyncBackgroundService : BackgroundService
                break;
            }

+            // Overflow-safe without checked math: the validator caps ReconnectBackoffMax at 1 day
+            // and backoff never exceeds it, so doubling stays far below long.MaxValue ticks.
            backoff = TimeSpan.FromTicks(Math.Min(backoff.Ticks * 2, _options.ReconnectBackoffMax.Ticks));
        }
    }
@@ -148,6 +148,31 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
        await bg.StopAsync(CancellationToken.None);
    }

+    // Shipped variant: in-proc against GrpcSyncDuplex with a flow-control-stalled fake writer
+    // (WriteAsync never completes until its token fires) — pins requirement 2 (Send honors its
+    // CancellationToken even when transport-blocked) deterministically; a true stalled-gRPC-peer
+    // rig in TestHost cannot force flow-control blockage reliably.
+    [Fact]
+    public async Task Send_StalledOnFlowControl_CancelUnblocksPromptly()
+    {
+        using var cts = new CancellationTokenSource();
+        var (duplex, readerTask) = Replication.Internal.GrpcSyncDuplex.Create(
+            new HangingReader(), new HangingWriter(), cts.Token);
+
+        var send = duplex.Send(new SyncMessage(), cts.Token);
+        Assert.False(send.IsCompleted);
+
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        cts.Cancel();
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => send);
+        // The reader pump must also tear down: its fault travels into the inbox completion.
+        await readerTask.WaitAsync(TimeSpan.FromSeconds(5));
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            async () => await duplex.Inbox.Completion);
+        sw.Stop();
+        Assert.True(sw.Elapsed < TimeSpan.FromSeconds(3), $"teardown took {sw.Elapsed}");
+    }
+
    // ---- validator -------------------------------------------------------------------------

    [Fact]
@@ -194,10 +219,23 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
        Assert.True(result.Succeeded);
    }

+    [Fact]
+    public void Validator_ReconnectBackoffMax_CappedAtOneDay()
+    {
+        var atCeiling = new ReplicationOptionsValidator().Validate(
+            null, new ReplicationOptions { ReconnectBackoffMax = TimeSpan.FromDays(1) });
+        Assert.True(atCeiling.Succeeded);
+
+        var overCeiling = new ReplicationOptionsValidator().Validate(
+            null, new ReplicationOptions { ReconnectBackoffMax = TimeSpan.FromDays(1) + TimeSpan.FromTicks(1) });
+        Assert.True(overCeiling.Failed);
+        Assert.Contains(overCeiling.Failures, f => f.Contains("ReconnectBackoffMax", StringComparison.Ordinal));
+    }
+
    // ---- DI --------------------------------------------------------------------------------

    [Fact]
-    public void AddZbLocalDbReplication_PassiveConfig_NoHostedService()
+    public async Task AddZbLocalDbReplication_PassiveConfig_InitiatorIdles()
    {
        var passive = new ServiceCollection();
        passive.AddLogging();
@@ -207,9 +245,17 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
        var p1 = passive.BuildServiceProvider();
        _providers.Add(p1);

-        Assert.DoesNotContain(p1.GetServices<IHostedService>(), s => s is SyncBackgroundService);
        Assert.NotNull(p1.GetService<LocalDbSyncService>());

+        // The hosted service is always registered; with no PeerAddress it starts, idles, and
+        // completes without dialing anything.
+        var bg = p1.GetServices<IHostedService>().OfType<SyncBackgroundService>().Single();
+        await bg.StartAsync(CancellationToken.None);
+        await bg.ExecuteTask!.WaitAsync(TimeSpan.FromSeconds(5));
+        Assert.True(bg.ExecuteTask.IsCompletedSuccessfully);
+        Assert.Equal(0, bg.ConnectionAttempts);
+        await bg.StopAsync(CancellationToken.None);
+
        var initiator = new ServiceCollection();
        initiator.AddLogging();
        var initiatorCfg = ConfigFor(NewDbPath(), peerAddress: "http://localhost", apiKey: null);
@@ -323,6 +369,27 @@ public sealed class GrpcAdapterTests : IAsyncLifetime
        throw new TimeoutException("Condition not reached within " + timeout);
    }

+    private sealed class HangingReader : IAsyncStreamReader<SyncMessage>
+    {
+        public SyncMessage Current => throw new InvalidOperationException("no message");
+
+        public async Task<bool> MoveNext(CancellationToken cancellationToken)
+        {
+            await Task.Delay(Timeout.Infinite, cancellationToken);
+            return false;
+        }
+    }
+
+    private sealed class HangingWriter : IAsyncStreamWriter<SyncMessage>
+    {
+        public WriteOptions? WriteOptions { get; set; }
+
+        public Task WriteAsync(SyncMessage message) => Task.Delay(Timeout.Infinite);
+
+        public Task WriteAsync(SyncMessage message, CancellationToken cancellationToken) =>
+            Task.Delay(Timeout.Infinite, cancellationToken);
+    }
+
    private sealed class AuthCaptureSink
    {
        public volatile string? Authorization;