feat(localdb): telemetry meters + ISyncStatus

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts

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<IValidateOptions<ReplicationOptions>, 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<SyncStatus>(sp =>
+        {
+            var db = SyncSessionFactory.RequireSqlite(sp.GetRequiredService<ILocalDb>(), nameof(SyncStatus));
+            var options = sp.GetRequiredService<IOptions<ReplicationOptions>>().Value;
+            var store = new OplogStore(db, options);
+            return new SyncStatus { OplogBacklogProvider = store.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,
+                LastSyncUtcProvider = () => status.LastSyncUtc,
+            };
+        });
+
         services.AddSingleton<LocalDbSyncService>();
         services.AddHostedService<SyncBackgroundService>();
 

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/ISyncStatus.cs

@@ -0,0 +1,82 @@
+namespace ZB.MOM.WW.LocalDb.Replication;
+
+/// <summary>
+/// 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.
+/// </summary>
+public interface ISyncStatus
+{
+    /// <summary>True while a sync session is actively running (in <c>RunAsync</c>).</summary>
+    bool Connected { get; }
+
+    /// <summary>The peer node's identity once the handshake has completed, else null.</summary>
+    string? PeerNodeId { get; }
+
+    /// <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>Number of connection attempts the initiator has made this lifetime.</summary>
+    long ConnectionAttempts { get; }
+}
+
+/// <summary>
+/// The mutable <see cref="ISyncStatus"/> backing store: thread-safe setters written from the sync
+/// session and the adapters, read from arbitrary dashboard / scrape threads. <see cref="OplogBacklog"/>
+/// polls <see cref="OplogBacklogProvider"/> on read (the same cheap SELECT the depth gauge uses).
+/// </summary>
+internal sealed class SyncStatus : ISyncStatus
+{
+    private volatile int _connected;
+    private string? _peerNodeId;
+    private long _lastSyncUtcTicks = -1;
+    private long _connectionAttempts;
+
+    /// <summary>Supplies the live unacked backlog for <see cref="OplogBacklog"/>; null yields 0.</summary>
+    public Func<long>? 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);
+}

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<D
         return rows[0];
     }
 
+    /// <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.
+    /// </summary>
+    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()!;
+    }
+
     /// <summary>The highest oplog seq present, or 0 when the oplog is empty.</summary>
     public async Task<long> GetMaxSeqAsync(CancellationToken ct = default)
     {

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<DateTimeOffset> _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<DateTimeOffset>? utcNow = null)
+        Func<DateTimeOffset>? 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;
     }
 
     /// <summary>
@@ -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<OplogEntryRecord>(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

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<SyncSession>();
-        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<SnapshotStreamer>());
         session.SnapshotSender = streamer.SendAsync;

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;
+
+/// <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).
+/// </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";
+
+    private readonly Meter _meter;
+    private readonly Counter<long> _applied;
+    private readonly Counter<long> _conflictsDiscarded;
+    private readonly Counter<long> _deadLettered;
+    private readonly Counter<long> _reconnects;
+    private readonly Counter<long> _snapshots;
+    private readonly Func<DateTimeOffset> _utcNow;
+    private bool _disposed;
+
+    public LocalDbMetrics(Func<DateTimeOffset>? utcNow = null)
+    {
+        _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
+        _meter = new Meter(MeterName, "0.1.0");
+        _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.sync.lag.seconds", ObserveSyncLag, "s");
+    }
+
+    /// <summary>
+    /// Supplies the current unacked oplog backlog for the <c>localdb.oplog.depth</c> gauge. Invoked
+    /// only at collection time; a cheap synchronous SELECT is acceptable at that frequency.
+    /// </summary>
+    public Func<long>? OplogDepthProvider { get; set; }
+
+    /// <summary>Supplies the last successful sync time for the <c>localdb.sync.lag.seconds</c> gauge; null means no sync yet, and the gauge then reports no measurement.</summary>
+    public Func<DateTimeOffset?>? LastSyncUtcProvider { get; set; }
+
+    /// <summary>The backing meter — exposed so a test <see cref="MeterListener"/> can filter by instance reference rather than the process-shared meter name.</summary>
+    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<string, object?>("direction", direction));
+
+    private IEnumerable<Measurement<long>> ObserveOplogDepth()
+    {
+        var provider = OplogDepthProvider;
+        if (provider is null)
+            return [];
+        try
+        {
+            return [new Measurement<long>(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<Measurement<double>> ObserveSyncLag()
+    {
+        var last = LastSyncUtcProvider?.Invoke();
+        if (last is null)
+            return [];
+        var lag = (_utcNow() - last.Value).TotalSeconds;
+        if (lag < 0) lag = 0;
+        return [new Measurement<double>(lag)];
+    }
+
+    public void Dispose()
+    {
+        if (_disposed)
+            return;
+        _meter.Dispose();
+        _disposed = true;
+    }
+}

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<LocalDbSyncService> _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<ReplicationOptions> 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<ReplicationOptions> options, ILoggerFactory loggerFactory,
+        LocalDbMetrics? metrics = null, ISyncStatus? status = null)
     {
         _db = SyncSessionFactory.RequireSqlite(db, nameof(LocalDbSyncService));
         _options = options.Value;
         _loggerFactory = loggerFactory;
         _logger = loggerFactory.CreateLogger<LocalDbSyncService>();
+        _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
                 {

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<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; }
@@ -33,12 +35,18 @@ public sealed class SyncBackgroundService : BackgroundService
     /// <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)
+    // 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<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)
@@ -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
                     {

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<string> _paths = [];
+    private readonly List<IDisposable> _disposables = [];
+    private readonly List<IHost> _hosts = [];
+    private readonly List<ServiceProvider> _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<ILocalDb>();
+        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)");
+        var metrics = clientProvider.GetRequiredService<LocalDbMetrics>();
+        using var collector = new Collector(metrics.Meter);
+
+        using var cts = new CancellationTokenSource(RunTimeout);
+        await bg.StartAsync(cts.Token);
+
+        var serverDb = serverHost.Services.GetRequiredService<ILocalDb>();
+        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<ILocalDb>();
+        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<ILocalDb>();
+        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)");
+        var status = clientProvider.GetRequiredService<ISyncStatus>();
+
+        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<Side> 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<SyncMessage>();
+        var bToA = Channel.CreateUnbounded<SyncMessage>();
+        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<SyncMessage> ToSession, ChannelReader<SyncMessage> FromSession) ScriptedPeer()
+    {
+        var toSession = Channel.CreateUnbounded<SyncMessage>();
+        var fromSession = Channel.CreateUnbounded<SyncMessage>();
+        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<SyncMessage> toSession, ChannelReader<SyncMessage> 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<SyncMessage> NextAsync(ChannelReader<SyncMessage> reader, CancellationToken ct)
+    {
+        await reader.WaitToReadAsync(ct);
+        reader.TryRead(out var msg);
+        return msg!;
+    }
+
+    private static async Task<SyncMessage> NextOfCaseAsync(
+        ChannelReader<SyncMessage> 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<string, string?>
+        {
+            ["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<IHost> 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<GrpcChannel> 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<IHostedService>().OfType<SyncBackgroundService>().Single();
+        bg.ChannelFactory = channelFactory;
+        return (provider, bg);
+    }
+
+    private static async Task<List<(long Id, string? Sku, long? Qty)>> 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<long> 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<Task<bool>> 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<string, List<long>> _longs = new(StringComparer.Ordinal);
+        private readonly Dictionary<string, List<double>> _doubles = new(StringComparer.Ordinal);
+
+        public Collector(Meter meter)
+        {
+            _listener = new MeterListener
+            {
+                InstrumentPublished = (inst, l) =>
+                {
+                    if (ReferenceEquals(inst.Meter, meter))
+                        l.EnableMeasurementEvents(inst);
+                },
+            };
+            _listener.SetMeasurementEventCallback<long>((inst, val, _, _) =>
+            {
+                lock (_lock) Append(_longs, inst.Name, val);
+            });
+            _listener.SetMeasurementEventCallback<double>((inst, val, _, _) =>
+            {
+                lock (_lock) Append(_doubles, inst.Name, val);
+            });
+            _listener.Start();
+        }
+
+        private static void Append<T>(Dictionary<string, List<T>> 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<double> 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();
+    }
+}