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 new file mode 100644 index 0000000..61962d9 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/DependencyInjection/ReplicationServiceCollectionExtensions.cs @@ -0,0 +1,54 @@ +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Routing; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Options; + +namespace ZB.MOM.WW.LocalDb.Replication; + +/// DI + endpoint wiring for the 2-node gRPC replication engine. +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. + /// + public static IServiceCollection AddZbLocalDbReplication( + this IServiceCollection services, IConfiguration configuration) + { + // Idempotent: a second call must not re-bind options or double-register the service/host. + if (services.Any(d => d.ServiceType == typeof(LocalDbSyncService))) + return services; + + services.AddOptions() + .Bind(configuration.GetSection("LocalDb:Replication")) + .ValidateOnStart(); + services.TryAddEnumerable( + 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(); + + return services; + } + + /// + /// Maps the passive gRPC endpoint. The host must have called + /// AddGrpc() and set up routing. + /// + public static IEndpointRouteBuilder MapZbLocalDbSync(this IEndpointRouteBuilder endpoints) + { + endpoints.MapGrpcService(); + return endpoints; + } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs new file mode 100644 index 0000000..8aa64a3 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/GrpcSyncDuplex.cs @@ -0,0 +1,59 @@ +using System.Threading.Channels; +using Grpc.Core; +using ZB.MOM.WW.LocalDb.Contracts; + +namespace ZB.MOM.WW.LocalDb.Replication.Internal; + +/// +/// Adapts a gRPC duplex stream (an inbound reader + an outbound writer) to the transport-agnostic +/// a runs over. Both the passive service and the +/// initiating client build one of these over their respective stream halves. +/// +internal static class GrpcSyncDuplex +{ + /// + /// Wires into an unbounded inbox channel drained by a background loop, + /// and exposes as the duplex send. Returns the duplex plus the reader + /// loop's task so the caller can observe it. On clean stream end the inbox channel is completed; + /// on a reader fault it is completed WITH the exception — the session's handshake and receive + /// phases both await the inbox, so terminating it is what unblocks them when the call dies (they + /// would otherwise hang on a dead stream). + /// + public static (SyncDuplex Duplex, Task ReaderTask) Create( + IAsyncStreamReader reader, + IAsyncStreamWriter writer, + CancellationToken ct) + { + var inbox = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleWriter = true }); + var readerTask = PumpInboundAsync(reader, inbox.Writer, ct); + var duplex = new SyncDuplex + { + // A plain delegate is safe: the session's single writer task is the only caller of Send, + // so this stream writer never sees the concurrent writes IServerStreamWriter / + // IClientStreamWriter both prohibit. The gRPC WriteAsync(message, ct) overload honors the + // token even when flow-control blocked, satisfying SyncDuplex.Send's cancellation contract. + Send = (message, token) => writer.WriteAsync(message, token), + Inbox = inbox.Reader, + }; + return (duplex, readerTask); + } + + private static async Task PumpInboundAsync( + IAsyncStreamReader reader, ChannelWriter inbox, CancellationToken ct) + { + try + { + while (await reader.MoveNext(ct)) + inbox.TryWrite(reader.Current); + } + catch (Exception ex) + { + // Fault the inbox so the session's inbox awaiters observe the transport failure (a + // cancelled call surfaces here as OperationCanceledException). Swallowed on this task: + // the failure travels to the session through the channel, not by faulting the reader task. + inbox.TryComplete(ex); + return; + } + inbox.TryComplete(); + } +} 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 new file mode 100644 index 0000000..3f9970d --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSessionFactory.cs @@ -0,0 +1,20 @@ +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.LocalDb.Internal; + +namespace ZB.MOM.WW.LocalDb.Replication.Internal; + +/// +/// Shared construction of a and its dependencies (OplogStore + LwwApplier) +/// from a and the bound options. Both the service and client adapters use +/// this so they build an identical session; snapshot hooks are left null (Task 12 fills them in). +/// +internal static class SyncSessionFactory +{ + public static SyncSession Create(SqliteLocalDb db, ReplicationOptions options, ILoggerFactory loggerFactory) + { + var store = new OplogStore(db, options); + var applier = new LwwApplier(db); + var logger = loggerFactory.CreateLogger(); + return new SyncSession(db, store, applier, options, logger); + } +} 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 new file mode 100644 index 0000000..2772a1a --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbSyncService.cs @@ -0,0 +1,62 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Contracts; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +namespace ZB.MOM.WW.LocalDb.Replication; + +/// +/// The passive (server) side of replication: a gRPC service that accepts an inbound bidirectional +/// Sync stream and runs the symmetric over it. Registered as a +/// singleton so its single-stream guard spans concurrent calls (a node syncs with exactly one peer). +/// +public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase +{ + private readonly SqliteLocalDb _db; + private readonly ReplicationOptions _options; + private readonly ILoggerFactory _loggerFactory; + + // 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}."); + _options = options.Value; + _loggerFactory = loggerFactory; + } + + public override async Task Sync( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context) + { + if (Interlocked.CompareExchange(ref _active, 1, 0) != 0) + throw new RpcException(new Status(StatusCode.AlreadyExists, "a sync stream is already active")); + + try + { + var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, context.CancellationToken); + var session = SyncSessionFactory.Create(_db, _options, _loggerFactory); + await session.RunAsync(duplex, context.CancellationToken); + await readerTask; + } + catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested) + { + // Client disconnect / server shutdown: the normal end of a long-lived streaming RPC. + } + catch (LocalDbSchemaMismatchException ex) + { + throw new RpcException(new Status(StatusCode.FailedPrecondition, ex.Message)); + } + finally + { + Interlocked.Exchange(ref _active, 0); + } + } +} 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 new file mode 100644 index 0000000..4f5cb37 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ReplicationOptionsValidator.cs @@ -0,0 +1,41 @@ +using Microsoft.Extensions.Options; + +namespace ZB.MOM.WW.LocalDb.Replication; + +/// +/// Validates at startup, accumulating every failure into one result. +/// is only required for the initiator role, which the +/// validator cannot distinguish from a passive node, so its format is checked only when present. +/// +public sealed class ReplicationOptionsValidator : IValidateOptions +{ + public ValidateOptionsResult Validate(string? name, ReplicationOptions options) + { + var failures = new List(); + + if (!string.IsNullOrEmpty(options.PeerAddress) + && (!Uri.TryCreate(options.PeerAddress, UriKind.Absolute, out var uri) + || (uri.Scheme != Uri.UriSchemeHttp && uri.Scheme != Uri.UriSchemeHttps))) + failures.Add( + $"LocalDb:Replication:PeerAddress must be an absolute http/https URI; got '{options.PeerAddress}'."); + + if (options.FlushInterval <= TimeSpan.Zero) + failures.Add($"LocalDb:Replication:FlushInterval must be greater than zero; got {options.FlushInterval}."); + if (options.MaxBatchSize <= 0) + failures.Add($"LocalDb:Replication:MaxBatchSize must be greater than 0; got {options.MaxBatchSize}."); + if (options.MaxOplogRows <= 0) + failures.Add($"LocalDb:Replication:MaxOplogRows must be greater than 0; got {options.MaxOplogRows}."); + if (options.MaxOplogAge <= TimeSpan.Zero) + failures.Add($"LocalDb:Replication:MaxOplogAge must be greater than zero; got {options.MaxOplogAge}."); + if (options.TombstoneRetention <= TimeSpan.Zero) + 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}."); + if (options.MaxHlcDriftAhead <= TimeSpan.Zero) + failures.Add($"LocalDb:Replication:MaxHlcDriftAhead must be greater than zero; got {options.MaxHlcDriftAhead}."); + + return failures.Count > 0 + ? ValidateOptionsResult.Fail(failures) + : ValidateOptionsResult.Success; + } +} 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 new file mode 100644 index 0000000..fb1964e --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/SyncBackgroundService.cs @@ -0,0 +1,120 @@ +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Contracts; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +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. +/// +public sealed class SyncBackgroundService : BackgroundService +{ + private static readonly TimeSpan InitialBackoff = TimeSpan.FromSeconds(1); + // A session healthier than this resets backoff to the floor: the fault was transient, not a + // fast-failing dial, so the next reconnect should not start from an inflated delay. + private static readonly TimeSpan HealthySessionThreshold = TimeSpan.FromSeconds(60); + + private readonly SqliteLocalDb _db; + private readonly ReplicationOptions _options; + private readonly ILoggerFactory _loggerFactory; + private readonly ILogger _logger; + + /// Test seam: supplies the gRPC channel instead of dialing . A channel it returns is owned by the caller and not disposed here. + internal Func? ChannelFactory { get; set; } + + /// Number of connection attempts made this lifetime (one per reconnect loop iteration). + internal int ConnectionAttempts; + + 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}."); + _options = options.Value; + _loggerFactory = loggerFactory; + _logger = loggerFactory.CreateLogger(); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var backoff = InitialBackoff; + while (!stoppingToken.IsCancellationRequested) + { + Interlocked.Increment(ref ConnectionAttempts); + var lifetime = System.Diagnostics.Stopwatch.StartNew(); + Exception? fault = null; + GrpcChannel? ownedChannel = null; + + try + { + var channel = ChannelFactory is { } factory + ? factory() + : ownedChannel = GrpcChannel.ForAddress(_options.PeerAddress); + + var client = new LocalDbSync.LocalDbSyncClient(channel); + var headers = new Metadata(); + if (!string.IsNullOrEmpty(_options.ApiKey)) + headers.Add("Authorization", $"Bearer {_options.ApiKey}"); + + using var call = client.Sync(headers, cancellationToken: stoppingToken); + var (duplex, readerTask) = GrpcSyncDuplex.Create(call.ResponseStream, call.RequestStream, stoppingToken); + var session = SyncSessionFactory.Create(_db, _options, _loggerFactory); + + await session.RunAsync(duplex, stoppingToken); + await call.RequestStream.CompleteAsync(); + await readerTask; + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + fault = ex; + } + finally + { + lifetime.Stop(); + ownedChannel?.Dispose(); + } + + if (IsSchemaMismatch(fault)) + // A schema mismatch will not self-heal by reconnecting; log loudly but keep trying + // (an operator may fix the peer) — capped at the max backoff so it does not spin. + _logger.LogError(fault, "Replication to peer {Peer} rejected on a schema mismatch; retrying at max backoff.", _options.PeerAddress); + else if (fault is not null) + _logger.LogWarning(fault, "Replication session to peer {Peer} faulted; reconnecting.", _options.PeerAddress); + + if (lifetime.Elapsed > HealthySessionThreshold) + backoff = InitialBackoff; + if (IsSchemaMismatch(fault)) + backoff = _options.ReconnectBackoffMax; + + try + { + await Task.Delay(Jitter(backoff), stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + + backoff = TimeSpan.FromTicks(Math.Min(backoff.Ticks * 2, _options.ReconnectBackoffMax.Ticks)); + } + } + + private static bool IsSchemaMismatch(Exception? ex) => + ex is LocalDbSchemaMismatchException + || (ex is RpcException rpc && rpc.StatusCode == StatusCode.FailedPrecondition); + + private static TimeSpan Jitter(TimeSpan value) => + value * (1.0 + (Random.Shared.NextDouble() * 0.4 - 0.2)); +} 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 new file mode 100644 index 0000000..8fdf009 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs @@ -0,0 +1,345 @@ +using Grpc.Core; +using Grpc.Core.Interceptors; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using ZB.MOM.WW.LocalDb.Contracts; +using ZB.MOM.WW.LocalDb.Replication; + +namespace ZB.MOM.WW.LocalDb.Tests; + +public sealed class GrpcAdapterTests : IAsyncLifetime +{ + private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(30); + + private readonly List _paths = []; + private readonly List _hosts = []; + private readonly List _providers = []; + + public Task InitializeAsync() => Task.CompletedTask; + + public async Task DisposeAsync() + { + foreach (var provider in _providers) + await provider.DisposeAsync(); + foreach (var host in _hosts) + { + try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ } + host.Dispose(); + } + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); + foreach (var p in _paths) + { + if (File.Exists(p)) File.Delete(p); + if (File.Exists(p + "-wal")) File.Delete(p + "-wal"); + if (File.Exists(p + "-shm")) File.Delete(p + "-shm"); + } + } + + // ---- end-to-end wire tests ------------------------------------------------------------- + + [Fact] + public async Task EndToEnd_TwoDbs_InsertOnA_AppearsOnB() + { + var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null); + var serverDb = serverHost.Services.GetRequiredService(); + await serverDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'FROM_SERVER', 20)"); + var handler = serverHost.GetTestServer().CreateHandler(); + + var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: null, () => ChannelOver(handler)); + var clientDb = clientProvider.GetRequiredService(); + await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'FROM_CLIENT', 10)"); + + using var cts = new CancellationTokenSource(RunTimeout); + await bg.StartAsync(cts.Token); + + await WaitForAsync( + async () => (await ReadOrders(clientDb)).Count == 2 && (await ReadOrders(serverDb)).Count == 2, + RunTimeout); + + var clientRows = await ReadOrders(clientDb); + var serverRows = await ReadOrders(serverDb); + Assert.Equal(clientRows, serverRows); + Assert.Equal([(1L, "FROM_CLIENT", (long?)10), (2L, "FROM_SERVER", (long?)20)], clientRows); + + await bg.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task SecondConcurrentStream_Rejected() + { + var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null); + var serverDb = serverHost.Services.GetRequiredService(); + var handler = serverHost.GetTestServer().CreateHandler(); + + var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: null, () => ChannelOver(handler)); + var clientDb = clientProvider.GetRequiredService(); + await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)"); + + using var cts = new CancellationTokenSource(RunTimeout); + await bg.StartAsync(cts.Token); + + // Convergence proves the first stream's server handler is active and holding the single-stream guard. + await WaitForAsync(async () => (await ReadOrders(serverDb)).Count == 1, RunTimeout); + + var client = new LocalDbSync.LocalDbSyncClient(ChannelOver(handler)); + using var call2 = client.Sync(cancellationToken: cts.Token); + var ex = await Assert.ThrowsAsync(() => call2.ResponseStream.MoveNext(cts.Token)); + Assert.Equal(StatusCode.AlreadyExists, ex.StatusCode); + + await bg.StopAsync(CancellationToken.None); + } + + // Shipped variant: the channel factory throws on its first call, then returns a live channel. + // This deterministically drives the backoff/retry loop (no flaky server restart in TestHost) + // and proves convergence after recovery. + [Fact] + public async Task ClientReconnects_AfterTransientFault() + { + var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null); + var serverDb = serverHost.Services.GetRequiredService(); + await serverDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'S', 20)"); + var handler = serverHost.GetTestServer().CreateHandler(); + + var calls = 0; + GrpcChannel Factory() + { + if (Interlocked.Increment(ref calls) == 1) + throw new InvalidOperationException("transient dial failure"); + return ChannelOver(handler); + } + + var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: null, Factory); + var clientDb = clientProvider.GetRequiredService(); + await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)"); + + using var cts = new CancellationTokenSource(RunTimeout); + await bg.StartAsync(cts.Token); + + await WaitForAsync( + async () => (await ReadOrders(clientDb)).Count == 2 && (await ReadOrders(serverDb)).Count == 2, + RunTimeout); + Assert.True(bg.ConnectionAttempts >= 2, $"expected >= 2 connection attempts, got {bg.ConnectionAttempts}"); + + await bg.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task ApiKeyHeader_SentWhenConfigured() + { + var sink = new AuthCaptureSink(); + var serverHost = await BuildServerAsync(NewDbPath(), apiKey: "test-key", sink: sink); + var handler = serverHost.GetTestServer().CreateHandler(); + + var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: "test-key", () => ChannelOver(handler)); + var clientDb = clientProvider.GetRequiredService(); + await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)"); + + using var cts = new CancellationTokenSource(RunTimeout); + await bg.StartAsync(cts.Token); + + await WaitForAsync(() => Task.FromResult(sink.Authorization is not null), RunTimeout); + Assert.Equal("Bearer test-key", sink.Authorization); + + await bg.StopAsync(CancellationToken.None); + } + + // ---- validator ------------------------------------------------------------------------- + + [Fact] + public void Validator_InvalidPeerAddress_Fails() + { + var result = new ReplicationOptionsValidator().Validate( + null, new ReplicationOptions { PeerAddress = "not-a-uri" }); + + Assert.True(result.Failed); + Assert.Contains(result.Failures, f => f.Contains("PeerAddress", StringComparison.Ordinal)); + } + + [Fact] + public void Validator_NonPositiveIntervals_Fail() + { + var result = new ReplicationOptionsValidator().Validate(null, new ReplicationOptions + { + FlushInterval = TimeSpan.Zero, + MaxBatchSize = 0, + MaxOplogRows = 0, + MaxOplogAge = TimeSpan.Zero, + TombstoneRetention = TimeSpan.Zero, + ReconnectBackoffMax = TimeSpan.Zero, + MaxHlcDriftAhead = TimeSpan.Zero, + }); + + Assert.True(result.Failed); + Assert.Contains(result.Failures, f => f.Contains("FlushInterval", StringComparison.Ordinal)); + Assert.Contains(result.Failures, f => f.Contains("MaxBatchSize", StringComparison.Ordinal)); + Assert.Contains(result.Failures, f => f.Contains("MaxOplogRows", StringComparison.Ordinal)); + Assert.Contains(result.Failures, f => f.Contains("MaxOplogAge", StringComparison.Ordinal)); + Assert.Contains(result.Failures, f => f.Contains("TombstoneRetention", StringComparison.Ordinal)); + Assert.Contains(result.Failures, f => f.Contains("ReconnectBackoffMax", StringComparison.Ordinal)); + Assert.Contains(result.Failures, f => f.Contains("MaxHlcDriftAhead", StringComparison.Ordinal)); + } + + [Fact] + public void Validator_EmptyPeerAddress_Allowed() + { + // A passive node leaves PeerAddress empty; the validator can't know the role, so it only + // checks format when present. All other defaults are valid. + var result = new ReplicationOptionsValidator().Validate(null, new ReplicationOptions { PeerAddress = "" }); + + Assert.True(result.Succeeded); + } + + // ---- DI -------------------------------------------------------------------------------- + + [Fact] + public void AddZbLocalDbReplication_PassiveConfig_NoHostedService() + { + var passive = new ServiceCollection(); + passive.AddLogging(); + var passiveCfg = ConfigFor(NewDbPath(), peerAddress: null, apiKey: null); + passive.AddZbLocalDb(passiveCfg, OnReady); + passive.AddZbLocalDbReplication(passiveCfg); + var p1 = passive.BuildServiceProvider(); + _providers.Add(p1); + + Assert.DoesNotContain(p1.GetServices(), s => s is SyncBackgroundService); + Assert.NotNull(p1.GetService()); + + var initiator = new ServiceCollection(); + initiator.AddLogging(); + var initiatorCfg = ConfigFor(NewDbPath(), peerAddress: "http://localhost", apiKey: null); + initiator.AddZbLocalDb(initiatorCfg, OnReady); + initiator.AddZbLocalDbReplication(initiatorCfg); + var p2 = initiator.BuildServiceProvider(); + _providers.Add(p2); + + Assert.Contains(p2.GetServices(), s => s is SyncBackgroundService); + } + + // ---- harness --------------------------------------------------------------------------- + + private string NewDbPath() + { + var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + _paths.Add(path); + return path; + } + + private static void OnReady(ILocalDb db) + { + using var conn = db.CreateConnection(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = "CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"; + cmd.ExecuteNonQuery(); + db.RegisterReplicated("orders"); + } + + private static IConfiguration ConfigFor(string path, string? peerAddress, string? apiKey) + { + var dict = new Dictionary + { + ["LocalDb:Path"] = path, + ["LocalDb:Replication:FlushInterval"] = "00:00:00.020", + }; + if (peerAddress is not null) dict["LocalDb:Replication:PeerAddress"] = peerAddress; + if (apiKey is not null) dict["LocalDb:Replication:ApiKey"] = apiKey; + return new ConfigurationBuilder().AddInMemoryCollection(dict).Build(); + } + + private static GrpcChannel ChannelOver(HttpMessageHandler handler) => + GrpcChannel.ForAddress("http://localhost", new GrpcChannelOptions { HttpHandler = handler }); + + private async Task BuildServerAsync(string path, string? apiKey, AuthCaptureSink? sink) + { + var config = ConfigFor(path, peerAddress: null, apiKey: apiKey); + var host = await new HostBuilder() + .ConfigureWebHost(web => + { + web.UseTestServer(); + web.ConfigureServices(services => + { + services.AddRouting(); + if (sink is not null) + { + services.AddSingleton(sink); + services.AddGrpc(o => o.Interceptors.Add()); + } + else + { + services.AddGrpc(); + } + + services.AddZbLocalDb(config, OnReady); + services.AddZbLocalDbReplication(config); + }); + web.Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapZbLocalDbSync()); + }); + }) + .StartAsync(); + _hosts.Add(host); + return host; + } + + private (ServiceProvider Provider, SyncBackgroundService Bg) BuildClient( + string path, string peerAddress, string? apiKey, Func channelFactory) + { + var config = ConfigFor(path, peerAddress, apiKey); + var services = new ServiceCollection(); + services.AddLogging(); + services.AddZbLocalDb(config, OnReady); + services.AddZbLocalDbReplication(config); + var provider = services.BuildServiceProvider(); + _providers.Add(provider); + + var bg = provider.GetServices().OfType().Single(); + bg.ChannelFactory = channelFactory; + return (provider, bg); + } + + private static async Task> ReadOrders(ILocalDb db) + { + var rows = await db.QueryAsync( + "SELECT id, sku, qty FROM orders ORDER BY id", + x => (x.GetInt64(0), x.IsDBNull(1) ? null : x.GetString(1), (long?)(x.IsDBNull(2) ? null : x.GetInt64(2)))); + return rows.ToList(); + } + + private static async Task WaitForAsync(Func> predicate, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await predicate()) return; + await Task.Delay(20); + } + throw new TimeoutException("Condition not reached within " + timeout); + } + + private sealed class AuthCaptureSink + { + public volatile string? Authorization; + } + + private sealed class AuthCaptureInterceptor(AuthCaptureSink sink) : Interceptor + { + public override Task DuplexStreamingServerHandler( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context, + DuplexStreamingServerMethod continuation) + { + var auth = context.RequestHeaders.GetValue("authorization"); + if (auth is not null) + sink.Authorization = auth; + return continuation(requestStream, responseStream, context); + } + } +}