fix(historian): latch the recorder's dropped-kind Debug log per run; pin the outbox.unreadable gauge

The DroppedUnsupportedKind counter still counts every value; only the Debug
line is latched (cleared by the next captured value), so a Gateway-backend
comms loss no longer logs one line per tag.

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

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs

@@ -196,6 +196,12 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
     /// the first item that ingests cleanly — see <see cref="IngestAsync"/>.</summary>
     private bool _ingestFaultLogged;
 
+    /// <summary>Whether the current run of unsupported-kind drops has already been logged, so a backend that
+    /// drops a whole class of value (every comms-loss marker on the Gateway backend, say) logs once per run rather
+    /// than once per value. The <see cref="_droppedUnsupportedKind"/> counter still counts every value. Cleared by
+    /// the next value that is captured.</summary>
+    private bool _droppedKindLogged;
+
     /// <summary>Last observed state of <see cref="_captureGate"/>, so a change is logged once rather
     /// than on every value. Seeded open: a gate that starts closed logs its first suppression.</summary>
     private bool _captureGateOpen = true;
@@ -553,10 +559,18 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
             // (e.g. String/Empty on the gateway's numeric-only SQL analog path) — drop and meter. No value
             // content is logged.
             _droppedUnsupportedKind++;
-            _log.Debug("ContinuousHistorization: dropped a value of a kind the historian backend does not support.");
+            if (!_droppedKindLogged)
+            {
+                _droppedKindLogged = true;
+                _log.Debug("ContinuousHistorization: dropped a value of a kind the historian backend does not support. "
+                    + "Further drops are only counted (DroppedUnsupportedKind) until a value is captured.");
+            }
+
             return;
         }
 
+        _droppedKindLogged = false;
+
         var timestampUtc = DateTime.SpecifyKind(msg.TimestampUtc, DateTimeKind.Utc);
 
         entries ??= new List<HistorizationOutboxEntry>();

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationMetricsTests.cs

@@ -0,0 +1,58 @@
+using System.Diagnostics.Metrics;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Observability;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
+
+/// <summary>
+///     The <c>otopcua.historian.outbox.unreadable</c> gauge reports the bound outbox's
+///     <see cref="IHistorizationOutbox.UnreadableCount"/>. The listener is filtered by the instrument INSTANCE (the
+///     shared <see cref="OtOpcUaTelemetry.Meter"/> object plus the instrument name), never by meter name alone, so a
+///     parallel test's measurements on another meter of the same name cannot reach it; and the bound outbox reports a
+///     distinctive count, so the asserted value can only have come from this test's outbox.
+/// </summary>
+public sealed class ContinuousHistorizationMetricsTests
+{
+    private const string GaugeName = "otopcua.historian.outbox.unreadable";
+
+    [Fact]
+    public void The_unreadable_gauge_reports_the_bound_outboxs_unreadable_count()
+    {
+        const long distinctive = 7_331_919L;
+        ContinuousHistorizationMetrics.BindOutbox(new UnreadableOutbox(distinctive));
+
+        var observed = new List<long>();
+        using var listener = new MeterListener
+        {
+            InstrumentPublished = (instrument, l) =>
+            {
+                if (ReferenceEquals(instrument.Meter, OtOpcUaTelemetry.Meter) && instrument.Name == GaugeName)
+                    l.EnableMeasurementEvents(instrument);
+            },
+        };
+        listener.SetMeasurementEventCallback<long>((instrument, value, _, _) =>
+        {
+            if (instrument.Name == GaugeName) observed.Add(value);
+        });
+        listener.Start();
+        listener.RecordObservableInstruments();
+
+        observed.ShouldBe([distinctive]);
+    }
+
+    private sealed class UnreadableOutbox(long unreadable) : IHistorizationOutbox
+    {
+        public long DroppedCount => 0;
+        public long UnreadableCount => unreadable;
+        public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => ValueTask.CompletedTask;
+        public ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct) => ValueTask.CompletedTask;
+        public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct) =>
+            ValueTask.FromResult<IReadOnlyList<HistorizationOutboxEntry>>([]);
+        public ValueTask RemoveAsync(Guid id, CancellationToken ct) => ValueTask.CompletedTask;
+        public ValueTask<int> CountAsync(CancellationToken ct) => ValueTask.FromResult(0);
+        public void Dispose() { }
+    }
+}