From 4e25f4f821797b7d2e557d30a6c2356db6a4eccd Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 17:48:10 -0400 Subject: [PATCH] fix(historian): append and alarm loops capture their lifetime token up front (lmxopcua#746) Both off-mailbox loops read _lifetime.Token per operation. PostStop disposes that source after the flush budget, and a disposed source's Token throws, so an operation started past the budget would have logged a spurious failure. Each loop now captures the token when it starts. Also rewrites the recorder's stale "capture Self before the await" note for the enqueue path. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- .../ContinuousHistorizationRecorder.cs | 24 +++++++------------ .../Historian/HistorianAdapterActor.cs | 10 ++++---- 2 files changed, 15 insertions(+), 19 deletions(-) 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 2943ae4c..debcda8a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -464,19 +464,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// A task that completes when the burst has been appended (or dropped). private async Task IngestAsync(IReadOnlyList items) { - // Capture Self BEFORE the await below. Self reads ActorBase.Context, and the append is awaited with - // ConfigureAwait(false) — which drops Akka's ActorTaskScheduler synchronization context, so a - // continuation that actually resumes off the actor thread finds NO ActorContext and throws - // NotSupportedException. This is the same idiom the GetStatus handler applies to Sender. - // - // It only bites when the append genuinely YIELDS: the production FasterLog outbox fsyncs per group - // commit (PerEntry commit mode), so it does; a fake outbox returning ValueTask.CompletedTask - // completes synchronously, the continuation runs inline with the context still installed, and the - // bug is invisible. That is exactly how it reached production — see the yielding fake in - // ContinuousHistorizationRecorderTests. Live-caught on the docker-dev rig, where the - // throw escaped the handler into supervision: the actor restarted, and because production seeds an - // EMPTY ref set (the deployed set arrives later via UpdateHistorizedRefs), the restart silently - // dropped every historized ref — so historization stopped dead after the FIRST value. + // Never touch Self (or Context) after the await below: Self reads ActorBase.Context, and the enqueue is awaited + // with ConfigureAwait(false), which drops Akka's ActorTaskScheduler synchronization context — a continuation that + // resumes off the actor thread finds NO ActorContext and throws NotSupportedException. That is how the first + // production recorder died (Gitea #491): the throw reached supervision, the restart discarded the EMPTY-seeded ref + // set, and historization stopped after the first value. The append loop holds its own captured Self. List? entries = null; for (int i = 0; i < items.Count; i++) { @@ -564,6 +556,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers private async Task AppendLoopAsync(IActorRef self) { var reader = _appendQueue.Reader; + // Captured now: PostStop disposes the source after the flush budget, and a disposed source's Token throws. + var lifetime = _lifetime.Token; while (true) { try @@ -586,10 +580,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers try { - await _outbox.AppendBatchAsync(merged, _lifetime.Token).ConfigureAwait(false); + await _outbox.AppendBatchAsync(merged, lifetime).ConfigureAwait(false); Interlocked.Add(ref _totalRecorded, merged.Count); } - catch (OperationCanceledException) when (_lifetime.IsCancellationRequested) + catch (OperationCanceledException) when (lifetime.IsCancellationRequested) { // Shutdown past the flush budget raced the append — not a fault. Interlocked.Add(ref _pendingAppends, -bursts); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs index a5629f94..bf2e408d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs @@ -182,8 +182,10 @@ public sealed class HistorianAdapterActor : ReceiveActor /// The single loop feeding the sink, in arrival order. Never throws: a rejected event is logged. private async Task DrainAsync() { + // Captured now: PostStop disposes the source after the flush budget, and a disposed source's Token throws. + var lifetime = _lifetime.Token; await foreach (var evt in _queue.Reader.ReadAllAsync(CancellationToken.None).ConfigureAwait(false)) - await EnqueueAsync(evt).ConfigureAwait(false); + await EnqueueAsync(evt, lifetime).ConfigureAwait(false); } /// Caches this node's from a cluster redundancy snapshot so the @@ -205,13 +207,13 @@ public sealed class HistorianAdapterActor : ReceiveActor } } - private async Task EnqueueAsync(AlarmHistorianEvent evt) + private async Task EnqueueAsync(AlarmHistorianEvent evt, CancellationToken lifetime) { try { - await _sink.EnqueueAsync(evt, _lifetime.Token).ConfigureAwait(false); + await _sink.EnqueueAsync(evt, lifetime).ConfigureAwait(false); } - catch (OperationCanceledException) when (_lifetime.IsCancellationRequested) + catch (OperationCanceledException) when (lifetime.IsCancellationRequested) { // Stopped past the flush budget. }