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. }