feat(localdb): gRPC sync adapters, background service, replication DI

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
2026-07-17 23:01:29 -04:00
parent 183901e8ac
commit a64fc26391
7 changed files with 701 additions and 0 deletions
@@ -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;
+
+/// <summary>DI + endpoint wiring for the 2-node gRPC replication engine.</summary>
+public static class ReplicationServiceCollectionExtensions
+{
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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<ReplicationOptions>()
+            .Bind(configuration.GetSection("LocalDb:Replication"))
+            .ValidateOnStart();
+        services.TryAddEnumerable(
+            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;
+    }
+
+    /// <summary>
+    /// Maps the passive <see cref="LocalDbSyncService"/> gRPC endpoint. The host must have called
+    /// <c>AddGrpc()</c> and set up routing.
+    /// </summary>
+    public static IEndpointRouteBuilder MapZbLocalDbSync(this IEndpointRouteBuilder endpoints)
+    {
+        endpoints.MapGrpcService<LocalDbSyncService>();
+        return endpoints;
+    }
+}
@@ -0,0 +1,59 @@
+using System.Threading.Channels;
+using Grpc.Core;
+using ZB.MOM.WW.LocalDb.Contracts;
+
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// Adapts a gRPC duplex stream (an inbound reader + an outbound writer) to the transport-agnostic
+/// <see cref="SyncDuplex"/> a <see cref="SyncSession"/> runs over. Both the passive service and the
+/// initiating client build one of these over their respective stream halves.
+/// </summary>
+internal static class GrpcSyncDuplex
+{
+    /// <summary>
+    /// Wires <paramref name="reader"/> into an unbounded inbox channel drained by a background loop,
+    /// and exposes <paramref name="writer"/> 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).
+    /// </summary>
+    public static (SyncDuplex Duplex, Task ReaderTask) Create(
+        IAsyncStreamReader<SyncMessage> reader,
+        IAsyncStreamWriter<SyncMessage> writer,
+        CancellationToken ct)
+    {
+        var inbox = Channel.CreateUnbounded<SyncMessage>(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<SyncMessage> reader, ChannelWriter<SyncMessage> 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();
+    }
+}
@@ -0,0 +1,20 @@
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.LocalDb.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// Shared construction of a <see cref="SyncSession"/> and its dependencies (OplogStore + LwwApplier)
+/// from a <see cref="SqliteLocalDb"/> 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).
+/// </summary>
+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<SyncSession>();
+        return new SyncSession(db, store, applier, options, logger);
+    }
+}
@@ -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;
+
+/// <summary>
+/// The passive (server) side of replication: a gRPC service that accepts an inbound bidirectional
+/// <c>Sync</c> stream and runs the symmetric <see cref="SyncSession"/> over it. Registered as a
+/// singleton so its single-stream guard spans concurrent calls (a node syncs with exactly one peer).
+/// </summary>
+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<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}.");
+        _options = options.Value;
+        _loggerFactory = loggerFactory;
+    }
+
+    public override async Task Sync(
+        IAsyncStreamReader<SyncMessage> requestStream,
+        IServerStreamWriter<SyncMessage> 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);
+        }
+    }
+}
@@ -0,0 +1,41 @@
+using Microsoft.Extensions.Options;
+
+namespace ZB.MOM.WW.LocalDb.Replication;
+
+/// <summary>
+/// Validates <see cref="ReplicationOptions"/> at startup, accumulating every failure into one result.
+/// <see cref="ReplicationOptions.PeerAddress"/> is only required for the initiator role, which the
+/// validator cannot distinguish from a passive node, so its format is checked only when present.
+/// </summary>
+public sealed class ReplicationOptionsValidator : IValidateOptions<ReplicationOptions>
+{
+    public ValidateOptionsResult Validate(string? name, ReplicationOptions options)
+    {
+        var failures = new List<string>();
+
+        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;
+    }
+}
@@ -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;
+
+/// <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.
+/// </summary>
+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<SyncBackgroundService> _logger;
+
+    /// <summary>Test seam: supplies the gRPC channel instead of dialing <see cref="ReplicationOptions.PeerAddress"/>. A channel it returns is owned by the caller and not disposed here.</summary>
+    internal Func<GrpcChannel>? ChannelFactory { get; set; }
+
+    /// <summary>Number of connection attempts made this lifetime (one per reconnect loop iteration).</summary>
+    internal int ConnectionAttempts;
+
+    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}.");
+        _options = options.Value;
+        _loggerFactory = loggerFactory;
+        _logger = loggerFactory.CreateLogger<SyncBackgroundService>();
+    }
+
+    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));
+}