fix(historian): pending-appends gauge follows the append queue back to zero (lmxopcua#746)

otopcua.historization.recorder.pending_appends was written only from the
actor thread: when a burst was accepted (just after the increment, so
never 0) and on GetStatus, which nothing in production sends. Once ingest
stopped the gauge froze at the depth the last burst saw, so after a
backpressure episode it kept reporting a queue near the 64-burst cap that
had long drained.

The append loop now publishes the depth after every decrement
(ReportPendingAppends). The mailbox-depth gauge stays actor-thread only;
its description now says when it is sampled. New test (non-parallel
collection, the gauge is process-wide): the gauge reads 2 with one append
stuck and one queued, and 0 after the queue drains with no status
request. With the loop's report removed it fails (stays 2).

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

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

@@ -46,18 +46,30 @@ public static class ContinuousHistorizationMetrics
             "otopcua.historization.recorder.pending_appends",
             () => (long)Volatile.Read(ref _pendingAppends),
             unit: "{burst}",
-            description: "Ingest bursts queued for, or in, the recorder's durable append at its last status (backpressure nears at the in-flight cap).");
+            description: "Ingest bursts queued for, or in, the recorder's durable append (backpressure nears at the in-flight cap).");
         OtOpcUaTelemetry.Meter.CreateObservableGauge(
             "otopcua.historization.recorder.mailbox_depth",
             () => (long)Volatile.Read(ref _mailboxDepth),
             unit: "{message}",
-            description: "Messages waiting in the continuous-historization recorder's mailbox at its last status (-1 = unknown).");
+            description: "Messages waiting in the continuous-historization recorder's mailbox when it last accepted a burst or answered a status (-1 = unknown).");
     }
 
     private static int _pendingAppends;
     private static int _mailboxDepth = -1;
 
-    /// <summary>Publishes the recorder's append-queue and mailbox depth for the gauges (called on each status).</summary>
+    /// <summary>The value the pending-appends gauge reports now — test/diagnostic.</summary>
+    internal static int PendingAppendsGauge => Volatile.Read(ref _pendingAppends);
+
+    /// <summary>
+    ///     Publishes the recorder's append-queue depth alone. The append loop calls it after every append completes, so the
+    ///     gauge falls back to 0 when the queue drains: the actor-thread reports (<see cref="ReportRecorder"/>) run only when
+    ///     a burst is accepted or a status is asked — nothing asks for status in production — so without this the gauge
+    ///     froze at the depth seen by the last burst (never 0, and still near the cap after a backpressure episode ended).
+    /// </summary>
+    /// <param name="pendingAppends">Bursts queued for or in the durable append.</param>
+    public static void ReportPendingAppends(int pendingAppends) => Volatile.Write(ref _pendingAppends, pendingAppends);
+
+    /// <summary>Publishes the recorder's append-queue and mailbox depth for the gauges (on each accepted burst and status).</summary>
     /// <param name="pendingAppends">Bursts queued for or in the durable append.</param>
     /// <param name="mailboxDepth">Messages waiting in the recorder's mailbox (-1 = unknown).</param>
     public static void ReportRecorder(int pendingAppends, int mailboxDepth)

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

@@ -534,7 +534,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
             catch (Exception ex) when (ex is OperationCanceledException or ChannelClosedException)
             {
                 // Normal shutdown raced the enqueue — not a fault.
-                Interlocked.Decrement(ref _pendingAppends);
+                ContinuousHistorizationMetrics.ReportPendingAppends(Interlocked.Decrement(ref _pendingAppends));
             }
         }
     }
@@ -586,7 +586,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
             catch (OperationCanceledException) when (lifetime.IsCancellationRequested)
             {
                 // Shutdown past the flush budget raced the append — not a fault.
-                Interlocked.Add(ref _pendingAppends, -bursts);
+                ContinuousHistorizationMetrics.ReportPendingAppends(Interlocked.Add(ref _pendingAppends, -bursts));
                 return;
             }
             catch (Exception ex)
@@ -600,7 +600,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
                     "the remainder are lost.", ex.GetType().Name, merged.Count, bursts);
             }
 
-            Interlocked.Add(ref _pendingAppends, -bursts);
+            // Published here, not only from the actor thread: otherwise the gauge keeps the depth the last burst saw.
+            ContinuousHistorizationMetrics.ReportPendingAppends(Interlocked.Add(ref _pendingAppends, -bursts));
             // Sent even after a failed append: any durably-appended prefix should drain now rather than wait for the
             // steady timer. The DrainTick handler de-dups and honours an active backoff. Not while stopping (dead letters).
             if (!_stopping) self.Tell(DrainTick.Instance);

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

@@ -0,0 +1,50 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
+
+/// <summary>
+///     The recorder's process-wide gauges. Their backing values are static, so these tests run alone (the collection
+///     disables parallelization): any other recorder test running at the same time would overwrite what is asserted.
+/// </summary>
+[CollectionDefinition(Name, DisableParallelization = true)]
+public sealed class RecorderGaugeCollection
+{
+    /// <summary>The collection name.</summary>
+    public const string Name = "historization-recorder-gauges";
+}
+
+/// <summary>The <c>otopcua.historization.recorder.pending_appends</c> gauge tracks the append queue, not the last burst.</summary>
+[Collection(RecorderGaugeCollection.Name)]
+public sealed class ContinuousHistorizationRecorderGaugeTests : TestKit
+{
+    private static readonly DateTime T0 = new(2026, 11, 9, 8, 0, 0, DateTimeKind.Utc);
+    private static readonly TimeSpan Wait = TimeSpan.FromSeconds(10);
+
+    private static VirtualTagActor.DependencyValueBatch Burst(int value) =>
+        new([new VirtualTagActor.DependencyValueChanged("A", (double)value, T0.AddSeconds(value))]);
+
+    [Fact]
+    public async Task The_pending_appends_gauge_returns_to_zero_when_the_queue_drains_without_a_status_request()
+    {
+        var log = new EventLog();
+        var outbox = new LoggingOutbox(log) { AppendGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) };
+        var rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            CreateTestProbe().Ref, new LoggingPassWriter(log), outbox, historizedRefs: ["A"], drainInterval: TimeSpan.FromSeconds(30)));
+
+        rec.Tell(Burst(0));
+        await outbox.AppendEntered.Task.WaitAsync(Wait);
+        rec.Tell(Burst(1));
+        // The second burst is accepted while the first append is stuck: the gauge shows the queue.
+        await AwaitAssertAsync(() => ContinuousHistorizationMetrics.PendingAppendsGauge.ShouldBe(2), Wait);
+
+        // No GetStatus from here on (production never asks one): the append loop itself must publish the drained depth.
+        outbox.AppendGate.SetResult();
+        await AwaitAssertAsync(() => outbox.Appended.Count.ShouldBe(2), Wait);
+        await AwaitAssertAsync(() => ContinuousHistorizationMetrics.PendingAppendsGauge.ShouldBe(0), Wait);
+    }
+}