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 de54e1c..59f0e80 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 @@ -4,6 +4,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Options; +using ZB.MOM.WW.LocalDb.Replication.Internal; namespace ZB.MOM.WW.LocalDb.Replication; @@ -32,6 +33,30 @@ public static class ReplicationServiceCollectionExtensions services.TryAddEnumerable( ServiceDescriptor.Singleton, ReplicationOptionsValidator>()); + // Telemetry is optional but always registered here; the sessions/adapters hold nullable + // references and null-guard every call, so a consumer that never collects metrics is + // unaffected. The oplog-depth / lag / backlog observers poll a cheap synchronous store SELECT. + services.TryAddSingleton(sp => + { + var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService(), nameof(SyncStatus)); + var options = sp.GetRequiredService>().Value; + var store = new OplogStore(db, options); + return new SyncStatus { OplogBacklogProvider = store.GetOplogDepthSync }; + }); + services.TryAddSingleton(sp => sp.GetRequiredService()); + services.TryAddSingleton(sp => + { + var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService(), nameof(LocalDbMetrics)); + var options = sp.GetRequiredService>().Value; + var store = new OplogStore(db, options); + var status = sp.GetRequiredService(); + return new LocalDbMetrics + { + OplogDepthProvider = store.GetOplogDepthSync, + LastSyncUtcProvider = () => status.LastSyncUtc, + }; + }); + services.AddSingleton(); services.AddHostedService(); diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ISyncStatus.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ISyncStatus.cs new file mode 100644 index 0000000..d872675 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ISyncStatus.cs @@ -0,0 +1,82 @@ +namespace ZB.MOM.WW.LocalDb.Replication; + +/// +/// A live, read-only view of the replication engine's current state, registered as a singleton for +/// dashboards / health surfaces. Always available (empty/false on a passive or not-yet-connected +/// node); the adapters and the sync session update it as connections and batches come and go. +/// +public interface ISyncStatus +{ + /// True while a sync session is actively running (in RunAsync). + bool Connected { get; } + + /// The peer node's identity once the handshake has completed, else null. + string? PeerNodeId { get; } + + /// When the last inbound batch was applied / acked, else null if no sync has occurred. + DateTimeOffset? LastSyncUtc { get; } + + /// The current unacked oplog backlog (entries above the peer's last-acked watermark). + long OplogBacklog { get; } + + /// Number of connection attempts the initiator has made this lifetime. + long ConnectionAttempts { get; } +} + +/// +/// The mutable backing store: thread-safe setters written from the sync +/// session and the adapters, read from arbitrary dashboard / scrape threads. +/// polls on read (the same cheap SELECT the depth gauge uses). +/// +internal sealed class SyncStatus : ISyncStatus +{ + private volatile int _connected; + private string? _peerNodeId; + private long _lastSyncUtcTicks = -1; + private long _connectionAttempts; + + /// Supplies the live unacked backlog for ; null yields 0. + public Func? OplogBacklogProvider { get; set; } + + public bool Connected => _connected != 0; + + public string? PeerNodeId => Volatile.Read(ref _peerNodeId); + + public DateTimeOffset? LastSyncUtc + { + get + { + var ticks = Interlocked.Read(ref _lastSyncUtcTicks); + return ticks < 0 ? null : new DateTimeOffset(ticks, TimeSpan.Zero); + } + } + + public long OplogBacklog + { + get + { + var provider = OplogBacklogProvider; + if (provider is null) + return 0; + try + { + return provider(); + } + catch + { + return 0; + } + } + } + + public long ConnectionAttempts => Interlocked.Read(ref _connectionAttempts); + + public void SetConnected(bool value) => _connected = value ? 1 : 0; + + public void SetPeerNodeId(string peerNodeId) => Volatile.Write(ref _peerNodeId, peerNodeId); + + public void SetLastSyncUtc(DateTimeOffset value) => + Interlocked.Exchange(ref _lastSyncUtcTicks, value.UtcTicks); + + public void SetConnectionAttempts(long value) => Interlocked.Exchange(ref _connectionAttempts, value); +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs index 88a090f..5e47fde 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs @@ -79,6 +79,21 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func + /// The unacked backlog read synchronously on a dedicated short-lived connection, for the + /// metric-observation callbacks (gauge scrape / status poll). Cheap enough at that frequency; + /// mirrors 's query. May throw if the db is disposing — callers + /// on the observation path swallow and drop the sample. + /// + public long GetOplogDepthSync() + { + using var conn = db.CreateConnection(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = + "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)"; + return (long)cmd.ExecuteScalar()!; + } + /// The highest oplog seq present, or 0 when the oplog is empty. public async Task GetMaxSeqAsync(CancellationToken ct = default) { diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs index ce9b91b..2dc5932 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs @@ -31,6 +31,8 @@ internal sealed class SyncSession private readonly ReplicationOptions _options; private readonly ILogger _logger; private readonly Func _utcNow; + private readonly LocalDbMetrics? _metrics; + private readonly SyncStatus? _status; // Highest oplog seq handed to the wire this session; the pump reads strictly above it so a batch // is never re-sent before its ack (the reliable ordered stream delivers exactly once in-session). @@ -45,7 +47,7 @@ internal sealed class SyncSession public SyncSession( SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger, - Func? utcNow = null) + Func? utcNow = null, LocalDbMetrics? metrics = null, SyncStatus? status = null) { _db = db; _store = store; @@ -53,6 +55,8 @@ internal sealed class SyncSession _options = options; _logger = logger; _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); + _metrics = metrics; + _status = status; } /// @@ -140,6 +144,7 @@ internal sealed class SyncSession await _store.SetPeerNodeIdAsync(peerHandshake.NodeId, ct); PeerNodeId = peerHandshake.NodeId; + _status?.SetPeerNodeId(peerHandshake.NodeId); _sentThruSeq = peerHandshake.LastAppliedRemoteSeq; Interlocked.Exchange(ref _peerAckedSeq, peerHandshake.LastAppliedRemoteSeq); @@ -164,6 +169,7 @@ internal sealed class SyncSession if (SnapshotSender is null) throw new NotSupportedException("snapshot required but no snapshot sender configured"); var snapshotAsOfSeq = await SnapshotSender(dataSend, ct); + _metrics?.RecordSnapshot("sent"); // The snapshot already carries every row state up to as-of; skipping the pump past it // avoids re-sending tail deltas the peer would only discard via LWW. if (snapshotAsOfSeq > _sentThruSeq) @@ -269,6 +275,7 @@ internal sealed class SyncSession "Replication protocol error: SnapshotComplete with no active snapshot (missing SnapshotBegin)."); await SnapshotApplier!.OnCompleteAsync(msg.SnapshotComplete, ct); _snapshotReceiving = false; + _metrics?.RecordSnapshot("received"); break; default: throw new InvalidOperationException( @@ -288,6 +295,7 @@ internal sealed class SyncSession var driftLimitMs = nowMs + (long)_options.MaxHlcDriftAhead.TotalMilliseconds; var toApply = new List(batch.Entries.Count); var maxSeq = 0L; + var driftDeadLettered = 0; foreach (var proto in batch.Entries) { @@ -310,6 +318,7 @@ internal sealed class SyncSession if (_options.FailClosedOnDrift) { await DeadLetterDriftedAsync(entry, physicalMs, nowMs, now, ct); + driftDeadLettered++; continue; } } @@ -317,6 +326,10 @@ internal sealed class SyncSession } var result = await _applier.ApplyBatchAsync(toApply, ct); + _metrics?.RecordApplied(result.Applied); + _metrics?.RecordConflictsDiscarded(result.Discarded); + _metrics?.RecordDeadLettered(result.DeadLettered + driftDeadLettered); + _status?.SetLastSyncUtc(now); var ackThru = Math.Max(result.AppliedThruSeq, maxSeq); if (ackThru > result.AppliedThruSeq) // Dead-lettered entries carried the batch's top seqs; advance the durable remote 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 6ef7c55..702c916 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 @@ -20,12 +20,14 @@ internal static class SyncSessionFactory $"{consumer} requires the ILocalDb registered by AddZbLocalDb (a {nameof(SqliteLocalDb)}); " + $"got {db.GetType().FullName}."); - public static SyncSession Create(SqliteLocalDb db, ReplicationOptions options, ILoggerFactory loggerFactory) + public static SyncSession Create( + SqliteLocalDb db, ReplicationOptions options, ILoggerFactory loggerFactory, + LocalDbMetrics? metrics = null, SyncStatus? status = null) { var store = new OplogStore(db, options); var applier = new LwwApplier(db); var logger = loggerFactory.CreateLogger(); - var session = new SyncSession(db, store, applier, options, logger); + var session = new SyncSession(db, store, applier, options, logger, utcNow: null, metrics, status); var streamer = new SnapshotStreamer(db, store, options, loggerFactory.CreateLogger()); session.SnapshotSender = streamer.SendAsync; diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs new file mode 100644 index 0000000..dbc98d3 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs @@ -0,0 +1,104 @@ +using System.Diagnostics.Metrics; + +namespace ZB.MOM.WW.LocalDb.Replication; + +/// +/// The replication engine's OpenTelemetry instruments, under the single zb.mom.ww.localdb +/// meter. Optional throughout: the sessions and adapters hold a nullable reference and every call +/// is null-guarded, so replication runs identically whether or not this is registered. Instrument +/// names follow the family metric conventions (lower-case, dot-separated, seconds for durations). +/// +public sealed class LocalDbMetrics : IDisposable +{ + /// The single meter name every replication instrument is published under. + public const string MeterName = "zb.mom.ww.localdb"; + + private readonly Meter _meter; + private readonly Counter _applied; + private readonly Counter _conflictsDiscarded; + private readonly Counter _deadLettered; + private readonly Counter _reconnects; + private readonly Counter _snapshots; + private readonly Func _utcNow; + private bool _disposed; + + public LocalDbMetrics(Func? utcNow = null) + { + _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); + _meter = new Meter(MeterName, "0.1.0"); + _applied = _meter.CreateCounter("localdb.sync.applied"); + _conflictsDiscarded = _meter.CreateCounter("localdb.sync.conflicts_discarded"); + _deadLettered = _meter.CreateCounter("localdb.sync.dead_lettered"); + _reconnects = _meter.CreateCounter("localdb.sync.reconnects"); + _snapshots = _meter.CreateCounter("localdb.sync.snapshots"); + _meter.CreateObservableGauge("localdb.oplog.depth", ObserveOplogDepth); + _meter.CreateObservableGauge("localdb.sync.lag.seconds", ObserveSyncLag, "s"); + } + + /// + /// Supplies the current unacked oplog backlog for the localdb.oplog.depth gauge. Invoked + /// only at collection time; a cheap synchronous SELECT is acceptable at that frequency. + /// + public Func? OplogDepthProvider { get; set; } + + /// Supplies the last successful sync time for the localdb.sync.lag.seconds gauge; null means no sync yet, and the gauge then reports no measurement. + public Func? LastSyncUtcProvider { get; set; } + + /// The backing meter — exposed so a test can filter by instance reference rather than the process-shared meter name. + internal Meter Meter => _meter; + + public void RecordApplied(long count) + { + if (count > 0) _applied.Add(count); + } + + public void RecordConflictsDiscarded(long count) + { + if (count > 0) _conflictsDiscarded.Add(count); + } + + public void RecordDeadLettered(long count) + { + if (count > 0) _deadLettered.Add(count); + } + + public void RecordReconnect() => _reconnects.Add(1); + + public void RecordSnapshot(string direction) => + _snapshots.Add(1, new KeyValuePair("direction", direction)); + + private IEnumerable> ObserveOplogDepth() + { + var provider = OplogDepthProvider; + if (provider is null) + return []; + try + { + return [new Measurement(provider())]; + } + catch + { + // The observation SELECT can race a disposed db / closing connection; a scrape must + // never throw, so drop this sample rather than surface a partial value. + return []; + } + } + + private IEnumerable> ObserveSyncLag() + { + var last = LastSyncUtcProvider?.Invoke(); + if (last is null) + return []; + var lag = (_utcNow() - last.Value).TotalSeconds; + if (lag < 0) lag = 0; + return [new Measurement(lag)]; + } + + public void Dispose() + { + if (_disposed) + return; + _meter.Dispose(); + _disposed = true; + } +} 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 86f469a..52f4caf 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 @@ -18,16 +18,24 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase private readonly ReplicationOptions _options; private readonly ILoggerFactory _loggerFactory; private readonly ILogger _logger; + private readonly LocalDbMetrics? _metrics; + private readonly SyncStatus? _status; // 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) + // status is taken as the public ISyncStatus (SyncStatus is internal and cannot appear in a + // public signature); DI always supplies the concrete SyncStatus this cast recovers. + public LocalDbSyncService( + ILocalDb db, IOptions options, ILoggerFactory loggerFactory, + LocalDbMetrics? metrics = null, ISyncStatus? status = null) { _db = SyncSessionFactory.RequireSqlite(db, nameof(LocalDbSyncService)); _options = options.Value; _loggerFactory = loggerFactory; _logger = loggerFactory.CreateLogger(); + _metrics = metrics; + _status = status as SyncStatus; } public override async Task Sync( @@ -46,13 +54,15 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase // 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); + var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status); + _status?.SetConnected(true); try { await session.RunAsync(duplex, linkedCts.Token); } finally { + _status?.SetConnected(false); linkedCts.Cancel(); try { 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 bf1bf4b..6e3e777 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 @@ -26,6 +26,8 @@ public sealed class SyncBackgroundService : BackgroundService private readonly ReplicationOptions _options; private readonly ILoggerFactory _loggerFactory; private readonly ILogger _logger; + private readonly LocalDbMetrics? _metrics; + private readonly SyncStatus? _status; /// 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; } @@ -33,12 +35,18 @@ public sealed class SyncBackgroundService : BackgroundService /// Number of connection attempts made this lifetime (one per reconnect loop iteration). internal int ConnectionAttempts; - public SyncBackgroundService(ILocalDb db, IOptions options, ILoggerFactory loggerFactory) + // status is taken as the public ISyncStatus (SyncStatus is internal and cannot appear in a + // public signature); DI always supplies the concrete SyncStatus this cast recovers. + public SyncBackgroundService( + ILocalDb db, IOptions options, ILoggerFactory loggerFactory, + LocalDbMetrics? metrics = null, ISyncStatus? status = null) { _db = SyncSessionFactory.RequireSqlite(db, nameof(SyncBackgroundService)); _options = options.Value; _loggerFactory = loggerFactory; _logger = loggerFactory.CreateLogger(); + _metrics = metrics; + _status = status as SyncStatus; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -52,7 +60,11 @@ public sealed class SyncBackgroundService : BackgroundService var backoff = InitialBackoff; while (!stoppingToken.IsCancellationRequested) { - Interlocked.Increment(ref ConnectionAttempts); + 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; @@ -75,7 +87,8 @@ public sealed class SyncBackgroundService : BackgroundService // 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); + var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status); + _status?.SetConnected(true); try { await session.RunAsync(duplex, linkedCts.Token); @@ -83,6 +96,7 @@ public sealed class SyncBackgroundService : BackgroundService } finally { + _status?.SetConnected(false); linkedCts.Cancel(); try { diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/MetricsTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/MetricsTests.cs new file mode 100644 index 0000000..947b36c --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/MetricsTests.cs @@ -0,0 +1,492 @@ +using System.Diagnostics.Metrics; +using System.Threading.Channels; +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 Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.LocalDb.Contracts; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.LocalDb.Replication.Internal; + +namespace ZB.MOM.WW.LocalDb.Tests; + +public sealed class MetricsTests : IAsyncLifetime +{ + private const string OrdersSql = "CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"; + private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(30); + + private readonly List _paths = []; + private readonly List _disposables = []; + 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(); + } + foreach (var d in _disposables) + d.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"); + } + } + + // ---- in-memory duplex tests ----------------------------------------------------------- + + [Fact] + public async Task Counters_IncrementOnApply() + { + using var metricsB = new LocalDbMetrics(); + using var collector = new Collector(metricsB.Meter); + + var a = await NewSideAsync(); + var b = await NewSideAsync(metrics: metricsB); + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'FROM_A', 10)"); + + var (da, db) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(db, cts.Token); + + await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 1, RunTimeout); + + Assert.True(collector.Sum("localdb.sync.applied") > 0); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + [Fact] + public async Task DeadLetter_Counter_Increments() + { + using var metrics = new LocalDbMetrics(); + using var collector = new Collector(metrics.Meter); + + var a = await NewSideAsync(metrics: metrics); + var (duplex, toSession, fromSession) = ScriptedPeer(); + + using var cts = new CancellationTokenSource(RunTimeout); + var run = a.Session.RunAsync(duplex, cts.Token); + + await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); + + // A non-tombstone entry with null row_json is poison: the applier dead-letters it. + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch + { + Entries = { new OplogEntry + { + Seq = 1, TableName = "orders", PkJson = "{\"id\":1}", + Hlc = 5_000_000, NodeId = "peer", IsTombstone = false, + } }, + }, + }, cts.Token); + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + + Assert.Equal(1, collector.Sum("localdb.sync.dead_lettered")); + + cts.Cancel(); + await SwallowAsync(run); + } + + [Fact] + public async Task OplogDepth_Gauge_ReportsBacklog() + { + using var metrics = new LocalDbMetrics(); + using var collector = new Collector(metrics.Meter); + + var a = await NewSideAsync(); + var b = await NewSideAsync(); + metrics.OplogDepthProvider = a.Store.GetOplogDepthSync; + + const int rows = 4; + for (var i = 1; i <= rows; i++) + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'R', @id)", new { id = i }); + + // No session running yet: the gauge reflects the unacked oplog backlog. + collector.Clear(); + collector.Observe(); + Assert.Equal(rows, collector.LastLong("localdb.oplog.depth")); + + var (da, db) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(db, cts.Token); + + await WaitForAsync(async () => await OplogCount(a.Db) == 0, RunTimeout); + + collector.Clear(); + collector.Observe(); + Assert.Equal(0, collector.LastLong("localdb.oplog.depth")); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + [Fact] + public async Task SyncLag_Gauge_EmptyUntilFirstSync_ThenReports() + { + using var metrics = new LocalDbMetrics(); + using var collector = new Collector(metrics.Meter); + + DateTimeOffset? lastSync = null; + metrics.LastSyncUtcProvider = () => lastSync; + + // Before any sync: the gauge yields no measurement (last_sync unknown). + collector.Observe(); + Assert.Empty(collector.Doubles("localdb.sync.lag.seconds")); + + lastSync = DateTimeOffset.UtcNow; + collector.Observe(); + var lags = collector.Doubles("localdb.sync.lag.seconds"); + Assert.NotEmpty(lags); + Assert.True(lags[^1] >= 0, $"lag {lags[^1]} should be non-negative"); + Assert.True(lags[^1] < 60, $"lag {lags[^1]} should be small"); + } + + [Fact] + public async Task Metrics_NullSafe_NoMetricsRegistered() + { + // A session constructed with metrics=null and status=null must sync without an NRE. + var a = await NewSideAsync(); + var b = await NewSideAsync(); + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 10)"); + + var (da, db) = DuplexPair(); + using var cts = new CancellationTokenSource(RunTimeout); + var runA = a.Session.RunAsync(da, cts.Token); + var runB = b.Session.RunAsync(db, cts.Token); + + await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 1, RunTimeout); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + // ---- gRPC host tests ------------------------------------------------------------------ + + [Fact] + public async Task Reconnects_Counter_Increments() + { + var serverHost = await BuildServerAsync(NewDbPath()); + 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", Factory); + var clientDb = clientProvider.GetRequiredService(); + await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)"); + var metrics = clientProvider.GetRequiredService(); + using var collector = new Collector(metrics.Meter); + + using var cts = new CancellationTokenSource(RunTimeout); + await bg.StartAsync(cts.Token); + + var serverDb = serverHost.Services.GetRequiredService(); + await WaitForAsync(async () => (await ReadOrders(serverDb)).Count == 1, RunTimeout); + + Assert.True(bg.ConnectionAttempts >= 2, $"attempts {bg.ConnectionAttempts}"); + Assert.True(collector.Sum("localdb.sync.reconnects") >= 1, $"reconnects {collector.Sum("localdb.sync.reconnects")}"); + + await bg.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task SyncStatus_ReflectsSessionLifecycle() + { + var serverHost = await BuildServerAsync(NewDbPath()); + var serverDb = serverHost.Services.GetRequiredService(); + await serverDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'S', 20)"); + var handler = serverHost.GetTestServer().CreateHandler(); + + var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", () => ChannelOver(handler)); + var clientDb = clientProvider.GetRequiredService(); + await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)"); + var status = clientProvider.GetRequiredService(); + + Assert.False(status.Connected); + + using var cts = new CancellationTokenSource(RunTimeout); + await bg.StartAsync(cts.Token); + + await WaitForAsync(() => Task.FromResult(status.Connected), RunTimeout); + await WaitForAsync( + async () => (await ReadOrders(clientDb)).Count == 2 && (await ReadOrders(serverDb)).Count == 2, + RunTimeout); + + Assert.NotNull(status.PeerNodeId); + Assert.NotNull(status.LastSyncUtc); + + await bg.StopAsync(CancellationToken.None); + await WaitForAsync(() => Task.FromResult(!status.Connected), RunTimeout); + Assert.False(status.Connected); + } + + // ---- in-memory harness ---------------------------------------------------------------- + + private sealed record Side(SqliteLocalDb Db, OplogStore Store, LwwApplier Applier, SyncSession Session); + + private async Task NewSideAsync(LocalDbMetrics? metrics = null, SyncStatus? status = null) + { + var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + _paths.Add(path); + var db = new SqliteLocalDb(new LocalDbOptions { Path = path }); + _disposables.Add(db); + await db.ExecuteAsync(OrdersSql); + db.RegisterReplicated("orders"); + + var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) }; + var store = new OplogStore(db, options); + var applier = new LwwApplier(db); + var session = new SyncSession(db, store, applier, options, NullLogger.Instance, null, metrics, status); + return new Side(db, store, applier, session); + } + + private static (SyncDuplex A, SyncDuplex B) DuplexPair() + { + var aToB = Channel.CreateUnbounded(); + var bToA = Channel.CreateUnbounded(); + var a = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader }; + var b = new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader }; + return (a, b); + } + + private static (SyncDuplex Duplex, ChannelWriter ToSession, ChannelReader FromSession) ScriptedPeer() + { + var toSession = Channel.CreateUnbounded(); + var fromSession = Channel.CreateUnbounded(); + var duplex = new SyncDuplex { Send = (m, ct) => fromSession.Writer.WriteAsync(m, ct).AsTask(), Inbox = toSession.Reader }; + return (duplex, toSession.Writer, fromSession.Reader); + } + + private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId) + { + var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = 0 }; + foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal)) + h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest }); + return h; + } + + private static async Task DriveHandshakeAsync( + SqliteLocalDb sessionDb, ChannelWriter toSession, ChannelReader fromSession, CancellationToken ct) + { + await NextAsync(fromSession, ct); + await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct); + await NextAsync(fromSession, ct); + await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct); + } + + private static async Task NextAsync(ChannelReader reader, CancellationToken ct) + { + await reader.WaitToReadAsync(ct); + reader.TryRead(out var msg); + return msg!; + } + + private static async Task NextOfCaseAsync( + ChannelReader reader, SyncMessage.MsgOneofCase msgCase, CancellationToken ct) + { + SyncMessage msg; + do { msg = await NextAsync(reader, ct); } while (msg.MsgCase != msgCase); + return msg; + } + + private static async Task SwallowAsync(Task task) + { + try { await task; } + catch (OperationCanceledException) { } + } + + // ---- gRPC host 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 = OrdersSql; + cmd.ExecuteNonQuery(); + db.RegisterReplicated("orders"); + } + + private static IConfiguration ConfigFor(string path, string? peerAddress) + { + var dict = new Dictionary + { + ["LocalDb:Path"] = path, + ["LocalDb:Replication:FlushInterval"] = "00:00:00.020", + }; + if (peerAddress is not null) dict["LocalDb:Replication:PeerAddress"] = peerAddress; + 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) + { + var config = ConfigFor(path, peerAddress: null); + var host = await new HostBuilder() + .ConfigureWebHost(web => + { + web.UseTestServer(); + web.ConfigureServices(services => + { + services.AddRouting(); + 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, Func channelFactory) + { + var config = ConfigFor(path, peerAddress); + 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 OplogCount(ILocalDb db) + { + var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0)); + return r[0]; + } + + 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); + } + + // ---- metric collector ----------------------------------------------------------------- + + private sealed class Collector : IDisposable + { + private readonly MeterListener _listener; + private readonly object _lock = new(); + private readonly Dictionary> _longs = new(StringComparer.Ordinal); + private readonly Dictionary> _doubles = new(StringComparer.Ordinal); + + public Collector(Meter meter) + { + _listener = new MeterListener + { + InstrumentPublished = (inst, l) => + { + if (ReferenceEquals(inst.Meter, meter)) + l.EnableMeasurementEvents(inst); + }, + }; + _listener.SetMeasurementEventCallback((inst, val, _, _) => + { + lock (_lock) Append(_longs, inst.Name, val); + }); + _listener.SetMeasurementEventCallback((inst, val, _, _) => + { + lock (_lock) Append(_doubles, inst.Name, val); + }); + _listener.Start(); + } + + private static void Append(Dictionary> map, string name, T value) + { + if (!map.TryGetValue(name, out var list)) + { + list = []; + map[name] = list; + } + list.Add(value); + } + + public void Observe() => _listener.RecordObservableInstruments(); + + public long Sum(string name) + { + lock (_lock) return _longs.TryGetValue(name, out var l) ? l.Sum() : 0; + } + + public long? LastLong(string name) + { + lock (_lock) return _longs.TryGetValue(name, out var l) && l.Count > 0 ? l[^1] : null; + } + + public IReadOnlyList Doubles(string name) + { + lock (_lock) return _doubles.TryGetValue(name, out var l) ? l.ToList() : []; + } + + public void Clear() + { + lock (_lock) + { + _longs.Clear(); + _doubles.Clear(); + } + } + + public void Dispose() => _listener.Dispose(); + } +}