diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs index 61962d9..de54e1c 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs @@ -14,10 +14,10 @@ public static class ReplicationServiceCollectionExtensions /// Binds and validates from the "LocalDb:Replication" /// configuration section, registers the passive as a singleton /// (its single-stream guard must span calls), and registers the initiating - /// as a hosted service ONLY when a peer address is configured - /// (a passive node leaves it empty and never dials). Idempotent. Requires - /// to have been called for the - /// the adapters depend on. + /// as a hosted service — on a passive node (no + /// ) it starts and immediately idles. Idempotent. + /// Requires to have been called + /// for the the adapters depend on. /// public static IServiceCollection AddZbLocalDbReplication( this IServiceCollection services, IConfiguration configuration) @@ -33,11 +33,7 @@ public static class ReplicationServiceCollectionExtensions ServiceDescriptor.Singleton, ReplicationOptionsValidator>()); services.AddSingleton(); - - // 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(); + services.AddHostedService(); return services; } diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs index 3f9970d..7c46bf3 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs @@ -10,6 +10,16 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal; /// internal static class SyncSessionFactory { + /// + /// The engine binds to internals (NodeId, Clock, replicated-table + /// digests), so both adapters require the concrete type AddZbLocalDb registers behind ILocalDb. + /// + 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); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSyncService.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSyncService.cs index 2772a1a..86f469a 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSyncService.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSyncService.cs @@ -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 _logger; // 0 = idle, 1 = a stream is active. Interlocked-guarded across concurrent calls (singleton service). private int _active; public LocalDbSyncService(ILocalDb db, IOptions 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(); } public override async Task Sync( @@ -41,10 +40,29 @@ 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); - await readerTask; + 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) { diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs index 4f5cb37..e657f88 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs @@ -9,6 +9,9 @@ namespace ZB.MOM.WW.LocalDb.Replication; /// public sealed class ReplicationOptionsValidator : IValidateOptions { + // 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(); @@ -31,6 +34,8 @@ public sealed class ReplicationOptionsValidator : IValidateOptions 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}."); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/SyncBackgroundService.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/SyncBackgroundService.cs index fb1964e..bf1bf4b 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/SyncBackgroundService.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/SyncBackgroundService.cs @@ -12,7 +12,8 @@ namespace ZB.MOM.WW.LocalDb.Replication; /// /// The initiating (client) side of replication: a hosted service that dials the configured peer, /// runs the symmetric 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 ) it idles. /// public sealed class SyncBackgroundService : BackgroundService { @@ -34,10 +35,7 @@ public sealed class SyncBackgroundService : BackgroundService public SyncBackgroundService(ILocalDb db, IOptions 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(); @@ -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,12 +69,30 @@ 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); - await call.RequestStream.CompleteAsync(); - await readerTask; + 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) { @@ -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)); } } diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs index 8fdf009..9d8e182 100644 --- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs @@ -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(() => send); + // The reader pump must also tear down: its fault travels into the inbox completion. + await readerTask.WaitAsync(TimeSpan.FromSeconds(5)); + await Assert.ThrowsAnyAsync( + 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(), s => s is SyncBackgroundService); Assert.NotNull(p1.GetService()); + // The hosted service is always registered; with no PeerAddress it starts, idles, and + // completes without dialing anything. + var bg = p1.GetServices().OfType().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 + { + public SyncMessage Current => throw new InvalidOperationException("no message"); + + public async Task MoveNext(CancellationToken cancellationToken) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return false; + } + } + + private sealed class HangingWriter : IAsyncStreamWriter + { + 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;