fix(localdb): telemetry review fixes — meter naming, snapshot status, session-count Connected

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
2026-07-18 00:14:57 -04:00
parent be4ca76f5f
commit eb40cd8f29
9 changed files with 130 additions and 40 deletions
@@ -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<SyncStatus>(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<ILocalDb>(), nameof(SyncStatus));
+            var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService<ILocalDb>(), nameof(OplogStore));
            var options = sp.GetRequiredService<IOptions<ReplicationOptions>>().Value;
-            var store = new OplogStore(db, options);
-            return new SyncStatus { OplogBacklogProvider = store.GetOplogDepthSync };
+            return new OplogStore(db, options);
        });
+        services.TryAddSingleton<SyncStatus>(sp =>
+            new SyncStatus { OplogBacklogProvider = sp.GetRequiredService<OplogStore>().GetOplogDepthSync });
        services.TryAddSingleton<ISyncStatus>(sp => sp.GetRequiredService<SyncStatus>());
        services.TryAddSingleton<LocalDbMetrics>(sp =>
        {
-            var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService<ILocalDb>(), nameof(LocalDbMetrics));
-            var options = sp.GetRequiredService<IOptions<ReplicationOptions>>().Value;
-            var store = new OplogStore(db, options);
            var status = sp.GetRequiredService<SyncStatus>();
            return new LocalDbMetrics
            {
-                OplogDepthProvider = store.GetOplogDepthSync,
+                OplogDepthProvider = sp.GetRequiredService<OplogStore>().GetOplogDepthSync,
                LastSyncUtcProvider = () => status.LastSyncUtc,
            };
        });
@@ -16,8 +16,8 @@ public interface ISyncStatus
    /// <summary>When the last inbound batch was applied / acked, else null if no sync has occurred.</summary>
    DateTimeOffset? LastSyncUtc { get; }

-    /// <summary>The current unacked oplog backlog (entries above the peer's last-acked watermark).</summary>
-    long OplogBacklog { get; }
+    /// <summary>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".</summary>
+    long? OplogBacklog { get; }

    /// <summary>Number of connection attempts the initiator has made this lifetime.</summary>
    long ConnectionAttempts { get; }
@@ -30,15 +30,17 @@ public interface ISyncStatus
 /// </summary>
 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;

-    /// <summary>Supplies the live unacked backlog for <see cref="OplogBacklog"/>; null yields 0.</summary>
+    /// <summary>Supplies the live unacked backlog for <see cref="OplogBacklog"/>; null / a throwing poll yields a null (unknown) backlog.</summary>
    public Func<long>? 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);

@@ -81,9 +81,12 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D

    /// <summary>
    /// 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 <see cref="GetOplogDepthAsync"/>'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
+    /// <see cref="GetOplogDepthAsync"/>'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
+    /// <c>BusyTimeoutMs</c> — 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.
    /// </summary>
    public long GetOplogDepthSync()
    {
@@ -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(
@@ -3,15 +3,17 @@ using System.Diagnostics.Metrics;
 namespace ZB.MOM.WW.LocalDb.Replication;

 /// <summary>
-/// The replication engine's OpenTelemetry instruments, under the single <c>zb.mom.ww.localdb</c>
-/// 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
+/// <c>ZB.MOM.WW.LocalDb.Replication</c> 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).
 /// </summary>
 public sealed class LocalDbMetrics : IDisposable
 {
    /// <summary>The single meter name every replication instrument is published under.</summary>
-    public const string MeterName = "zb.mom.ww.localdb";
+    public const string MeterName = "ZB.MOM.WW.LocalDb.Replication";

    private readonly Meter _meter;
    private readonly Counter<long> _applied;
@@ -25,13 +27,13 @@ public sealed class LocalDbMetrics : IDisposable
    public LocalDbMetrics(Func<DateTimeOffset>? 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<long>("localdb.sync.applied");
        _conflictsDiscarded = _meter.CreateCounter<long>("localdb.sync.conflicts_discarded");
        _deadLettered = _meter.CreateCounter<long>("localdb.sync.dead_lettered");
        _reconnects = _meter.CreateCounter<long>("localdb.sync.reconnects");
        _snapshots = _meter.CreateCounter<long>("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<string, object?>("direction", direction));
+    public void RecordSnapshotSent() =>
+        _snapshots.Add(1, new KeyValuePair<string, object?>("direction", "sent"));
+
+    public void RecordSnapshotReceived() =>
+        _snapshots.Add(1, new KeyValuePair<string, object?>("direction", "received"));

    private IEnumerable<Measurement<long>> ObserveOplogDepth()
    {
@@ -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
                {
@@ -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
                    {
@@ -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<SqliteConnection> CreateConnectionAsync(CancellationToken ct)
@@ -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<ILocalDb>();
        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)");
        var status = clientProvider.GetRequiredService<ISyncStatus>();
+        var serverStatus = serverHost.Services.GetRequiredService<ISyncStatus>();

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