diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs index 09be12f5a..f8e275585 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationMetrics.cs +++ b/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; - /// Publishes the recorder's append-queue and mailbox depth for the gauges (called on each status). + /// The value the pending-appends gauge reports now — test/diagnostic. + internal static int PendingAppendsGauge => Volatile.Read(ref _pendingAppends); + + /// + /// 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 () 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). + /// + /// Bursts queued for or in the durable append. + public static void ReportPendingAppends(int pendingAppends) => Volatile.Write(ref _pendingAppends, pendingAppends); + + /// Publishes the recorder's append-queue and mailbox depth for the gauges (on each accepted burst and status). /// Bursts queued for or in the durable append. /// Messages waiting in the recorder's mailbox (-1 = unknown). public static void ReportRecorder(int pendingAppends, int mailboxDepth) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs index debcda8ac..10641dc81 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/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); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderGaugeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderGaugeTests.cs new file mode 100644 index 000000000..1510540a7 --- /dev/null +++ b/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; + +/// +/// 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. +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class RecorderGaugeCollection +{ + /// The collection name. + public const string Name = "historization-recorder-gauges"; +} + +/// The otopcua.historization.recorder.pending_appends gauge tracks the append queue, not the last burst. +[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); + } +}