From eb40cd8f2986bae0081e74a0cba46d88ce4f0d35 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 18 Jul 2026 00:14:57 -0400 Subject: [PATCH] =?UTF-8?q?fix(localdb):=20telemetry=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20meter=20naming,=20snapshot=20status,=20session-coun?= =?UTF-8?q?t=20Connected?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts --- .../ReplicationServiceCollectionExtensions.cs | 18 ++--- .../ISyncStatus.cs | 22 +++--- .../Internal/OplogStore.cs | 9 ++- .../Internal/SyncSession.cs | 6 +- .../LocalDbMetrics.cs | 23 ++++--- .../LocalDbSyncService.cs | 4 +- .../SyncBackgroundService.cs | 4 +- .../Internal/SqliteLocalDb.cs | 16 +++-- .../ZB.MOM.WW.LocalDb.Tests/MetricsTests.cs | 68 +++++++++++++++++++ 9 files changed, 130 insertions(+), 40 deletions(-) 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 59f0e80..f9ccba9 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 @@ -35,24 +35,24 @@ public static class ReplicationServiceCollectionExtensions // 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 => + // unaffected. One OplogStore instance backs both the status backlog and the depth gauge + // (it is stateless over the db, so sharing is free); its observers poll a cheap + // synchronous SELECT. + services.TryAddSingleton(sp => { - var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService(), nameof(SyncStatus)); + var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService(), nameof(OplogStore)); var options = sp.GetRequiredService>().Value; - var store = new OplogStore(db, options); - return new SyncStatus { OplogBacklogProvider = store.GetOplogDepthSync }; + return new OplogStore(db, options); }); + services.TryAddSingleton(sp => + new SyncStatus { OplogBacklogProvider = sp.GetRequiredService().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, + OplogDepthProvider = sp.GetRequiredService().GetOplogDepthSync, LastSyncUtcProvider = () => status.LastSyncUtc, }; }); 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 index d872675..e41e0d1 100644 --- 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 @@ -16,8 +16,8 @@ public interface ISyncStatus /// 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; } + /// The current unacked oplog backlog (entries above the peer's last-acked watermark), or null when unknown (no provider wired / the poll failed) — so a DB failure never masquerades as "0 backlog". + long? OplogBacklog { get; } /// Number of connection attempts the initiator has made this lifetime. long ConnectionAttempts { get; } @@ -30,15 +30,17 @@ public interface ISyncStatus /// internal sealed class SyncStatus : ISyncStatus { - private volatile int _connected; + // A node can run BOTH roles at once (initiator + passive server), so Connected tracks a count + // of active sessions rather than a boolean — one session ending must not clobber the other. + private int _activeSessions; private string? _peerNodeId; private long _lastSyncUtcTicks = -1; private long _connectionAttempts; - /// Supplies the live unacked backlog for ; null yields 0. + /// Supplies the live unacked backlog for ; null / a throwing poll yields a null (unknown) backlog. public Func? OplogBacklogProvider { get; set; } - public bool Connected => _connected != 0; + public bool Connected => Volatile.Read(ref _activeSessions) > 0; public string? PeerNodeId => Volatile.Read(ref _peerNodeId); @@ -51,27 +53,29 @@ internal sealed class SyncStatus : ISyncStatus } } - public long OplogBacklog + public long? OplogBacklog { get { var provider = OplogBacklogProvider; if (provider is null) - return 0; + return null; try { return provider(); } catch { - return 0; + return null; } } } public long ConnectionAttempts => Interlocked.Read(ref _connectionAttempts); - public void SetConnected(bool value) => _connected = value ? 1 : 0; + public void SessionStarted() => Interlocked.Increment(ref _activeSessions); + + public void SessionEnded() => Interlocked.Decrement(ref _activeSessions); public void SetPeerNodeId(string peerNodeId) => Volatile.Write(ref _peerNodeId, peerNodeId); 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 5e47fde..424f383 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 @@ -81,9 +81,12 @@ 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. + /// metric-observation callbacks (gauge scrape / status poll); mirrors + /// 's query. Deliberate tradeoff: synchronous I/O on the + /// metrics-collection thread is acceptable here because WAL-mode readers never block (or are + /// blocked by) writers, the COUNT is indexed-cheap, and the worst case is bounded by + /// BusyTimeoutMs — all at scrape frequency, not on any data path. May throw if the db + /// is disposing — callers on the observation path swallow and drop the sample. /// public long GetOplogDepthSync() { 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 2dc5932..23d66db 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 @@ -169,7 +169,8 @@ 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"); + _metrics?.RecordSnapshotSent(); + _status?.SetLastSyncUtc(_utcNow()); // 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) @@ -275,7 +276,8 @@ 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"); + _metrics?.RecordSnapshotReceived(); + _status?.SetLastSyncUtc(_utcNow()); break; default: throw new InvalidOperationException( 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 index dbc98d3..50ed31d 100644 --- 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 @@ -3,15 +3,17 @@ 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). +/// The replication engine's OpenTelemetry instruments, under the single +/// ZB.MOM.WW.LocalDb.Replication meter (family METRIC-CONVENTIONS §1: meters are named +/// after their root namespace). 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"; + public const string MeterName = "ZB.MOM.WW.LocalDb.Replication"; private readonly Meter _meter; private readonly Counter _applied; @@ -25,13 +27,13 @@ public sealed class LocalDbMetrics : IDisposable public LocalDbMetrics(Func? utcNow = null) { _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); - _meter = new Meter(MeterName, "0.1.0"); + _meter = new Meter(MeterName, typeof(LocalDbMetrics).Assembly.GetName().Version?.ToString(3)); _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.oplog.depth", ObserveOplogDepth, "1"); _meter.CreateObservableGauge("localdb.sync.lag.seconds", ObserveSyncLag, "s"); } @@ -64,8 +66,11 @@ public sealed class LocalDbMetrics : IDisposable public void RecordReconnect() => _reconnects.Add(1); - public void RecordSnapshot(string direction) => - _snapshots.Add(1, new KeyValuePair("direction", direction)); + public void RecordSnapshotSent() => + _snapshots.Add(1, new KeyValuePair("direction", "sent")); + + public void RecordSnapshotReceived() => + _snapshots.Add(1, new KeyValuePair("direction", "received")); private IEnumerable> ObserveOplogDepth() { 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 52f4caf..3155642 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 @@ -55,14 +55,14 @@ public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken); var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, linkedCts.Token); var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status); - _status?.SetConnected(true); + _status?.SessionStarted(); try { await session.RunAsync(duplex, linkedCts.Token); } finally { - _status?.SetConnected(false); + _status?.SessionEnded(); 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 6e3e777..429ec97 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 @@ -88,7 +88,7 @@ public sealed class SyncBackgroundService : BackgroundService 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?.SetConnected(true); + _status?.SessionStarted(); try { await session.RunAsync(duplex, linkedCts.Token); @@ -96,7 +96,7 @@ public sealed class SyncBackgroundService : BackgroundService } finally { - _status?.SetConnected(false); + _status?.SessionEnded(); linkedCts.Cancel(); try { diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs index abfd1e5..ffd390e 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs @@ -83,10 +83,18 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable public SqliteConnection CreateConnection() { var conn = new SqliteConnection(_connectionString); - conn.Open(); - ExecuteRaw(conn, PerConnectionPragmas()); - conn.CreateFunction("zb_hlc_next", () => _clock.Next()); - return conn; + try + { + conn.Open(); + ExecuteRaw(conn, PerConnectionPragmas()); + conn.CreateFunction("zb_hlc_next", () => _clock.Next()); + return conn; + } + catch + { + conn.Dispose(); + throw; + } } internal async Task CreateConnectionAsync(CancellationToken ct) 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 index 947b36c..61e99fb 100644 --- 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 @@ -163,6 +163,65 @@ public sealed class MetricsTests : IAsyncLifetime Assert.True(lags[^1] < 60, $"lag {lags[^1]} should be small"); } + [Fact] + public async Task SnapshotOnlySync_UpdatesLastSyncAndLag() + { + var statusA = new SyncStatus(); + var statusB = new SyncStatus(); + using var metricsB = new LocalDbMetrics { LastSyncUtcProvider = () => statusB.LastSyncUtc }; + using var collector = new Collector(metricsB.Meter); + + var a = await NewSideAsync(status: statusA); + var b = await NewSideAsync(metrics: metricsB, status: statusB); + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'SNAP', 10)"); + // A owes B a full snapshot; the pump then skips the tail deltas the snapshot covers, so + // B converges via the snapshot path alone (no delta batch ever reaches it). + await a.Store.SetNeedsSnapshotAsync(true, default); + + // Before any sync the lag gauge yields no measurement. + collector.Observe(); + Assert.Empty(collector.Doubles("localdb.sync.lag.seconds")); + + 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); + await WaitForAsync(() => Task.FromResult(statusB.LastSyncUtc is not null), RunTimeout); + + Assert.NotNull(statusB.LastSyncUtc); // receive path: SnapshotComplete + Assert.NotNull(statusA.LastSyncUtc); // send path: SnapshotSender returned + + 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"); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + [Fact] + public void BothRoles_ActiveSessions_ConnectedUntilLastEnds() + { + // A node running both roles holds two concurrent sessions; one ending must not clobber + // the other's Connected signal. + var status = new SyncStatus(); + Assert.False(status.Connected); + + status.SessionStarted(); + Assert.True(status.Connected); + status.SessionStarted(); + Assert.True(status.Connected); + + status.SessionEnded(); + Assert.True(status.Connected); + status.SessionEnded(); + Assert.False(status.Connected); + } + [Fact] public async Task Metrics_NullSafe_NoMetricsRegistered() { @@ -229,23 +288,29 @@ public sealed class MetricsTests : IAsyncLifetime var clientDb = clientProvider.GetRequiredService(); await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)"); var status = clientProvider.GetRequiredService(); + var serverStatus = serverHost.Services.GetRequiredService(); Assert.False(status.Connected); + Assert.False(serverStatus.Connected); using var cts = new CancellationTokenSource(RunTimeout); await bg.StartAsync(cts.Token); await WaitForAsync(() => Task.FromResult(status.Connected), RunTimeout); + await WaitForAsync(() => Task.FromResult(serverStatus.Connected), RunTimeout); await WaitForAsync( async () => (await ReadOrders(clientDb)).Count == 2 && (await ReadOrders(serverDb)).Count == 2, RunTimeout); Assert.NotNull(status.PeerNodeId); Assert.NotNull(status.LastSyncUtc); + Assert.NotNull(serverStatus.PeerNodeId); await bg.StopAsync(CancellationToken.None); await WaitForAsync(() => Task.FromResult(!status.Connected), RunTimeout); Assert.False(status.Connected); + await WaitForAsync(() => Task.FromResult(!serverStatus.Connected), RunTimeout); + Assert.False(serverStatus.Connected); } // ---- in-memory harness ---------------------------------------------------------------- @@ -265,6 +330,9 @@ public sealed class MetricsTests : IAsyncLifetime var store = new OplogStore(db, options); var applier = new LwwApplier(db); var session = new SyncSession(db, store, applier, options, NullLogger.Instance, null, metrics, status); + var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance); + session.SnapshotSender = streamer.SendAsync; + session.SnapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance); return new Side(db, store, applier, session); }