scadaproj/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/SyncBackgroundService.cs

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. Always
/// registered; on a passive node (no <see cref="ReplicationOptions.PeerAddress"/>) it idles.
/// </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;
    private readonly LocalDbMetrics? _metrics;
    private readonly SyncStatus? _status;

    /// <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;

    /// <summary>
    /// Creates the initiating sync host. <paramref name="status"/> is taken as the public
    /// <see cref="ISyncStatus"/> (SyncStatus is internal and cannot appear in a public signature);
    /// DI always supplies the concrete SyncStatus the ctor recovers by cast.
    /// </summary>
    public SyncBackgroundService(
        ILocalDb db, IOptions<ReplicationOptions> options, ILoggerFactory loggerFactory,
        LocalDbMetrics? metrics = null, ISyncStatus? status = null)
    {
        _db = SyncSessionFactory.RequireSqlite(db, nameof(SyncBackgroundService));
        _options = options.Value;
        _loggerFactory = loggerFactory;
        _logger = loggerFactory.CreateLogger<SyncBackgroundService>();
        _metrics = metrics;
        _status = status as SyncStatus;
    }

    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)
        {
            var attempt = Interlocked.Increment(ref ConnectionAttempts);
            _status?.SetConnectionAttempts(attempt);
            // Every attempt past the first is a reconnect (the initial dial is not).
            if (attempt > 1)
                _metrics?.RecordReconnect();
            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);
                // 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, _metrics, _status);
                _status?.SessionStarted();
                try
                {
                    await session.RunAsync(duplex, linkedCts.Token);
                    await call.RequestStream.CompleteAsync();
                }
                finally
                {
                    _status?.SessionEnded();
                    linkedCts.Cancel();
                    try
                    {
                        await readerTask;
                    }
                    catch (Exception ex)
                    {
                        _logger.LogDebug(ex, "Sync reader pump ended during teardown.");
                    }
                }
            }
            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;
            }

            // 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));
        }
    }

    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));
}