fix(historian): the recorder died on its first value — live gate for continuous historization (#491)
Ran #491's live gate against a real ZB.MOM.WW.HistorianGateway talking to a real AVEVA Historian, and it found a production-fatal defect on the first value it ever saw. ContinuousHistorizationRecorder.OnValueChangedAsync touched `Self` after awaiting IHistorizationOutbox.AppendAsync(...).ConfigureAwait(false). ConfigureAwait(false) discards Akka's ActorTaskScheduler synchronization context, so `Self` — which reads ActorBase.Context — threw NotSupportedException. The throw escaped into supervision, which RESTARTED the actor; because production seeds an EMPTY ref set (the deployed set arrives later via UpdateHistorizedRefs), the restart silently discarded every historized ref. Net effect: one value captured, then nothing, with no further error to show for it. Fixed by capturing `IActorRef self = Self` before the await — the idiom the GetStatus handler already applied to Sender. Thirteen unit tests missed it because InMemoryOutbox.AppendAsync returned ValueTask.CompletedTask: a completed ValueTask never yields, so the continuation ran inline with the context still installed. The production FasterLog outbox fsyncs per entry and does yield. The first regression test was itself vacuous — `await Task.Yield()` posts back to the CAPTURED context, so the append still completed on the actor thread and reverting the fix left the test green. The fake must complete on a POOL thread (Task.Delay(1).ConfigureAwait(false)), as a real fsync does; with that, reverting the fix reddens exactly one test. The rig could not run this gate at all: docker-dev carried ServerHistorian__* but no ContinuousHistorization__* keys, so the recorder could never spawn. Both centrals now carry them (default OFF — the recorder needs ContinuousHistorization:Enabled AND ServerHistorian:Enabled), with the outbox on the per-node named volume so it survives a container recreate. ServerHistorian__UseTls was hard-coded true and is now parameterized; a plaintext h2c gateway was unreachable without it. Gate results: outbox append, drain, WriteLiveValues accepted by a real historian, and restart convergence (capture resumes after a restart with no new deployment) all pass. The read-back hop is NOT proven — the LAN historian accepts writes but serves no reads under a "License violation: Database tag count exceeds license tag count", and the licensed corporate one needs the VPN. EnsureTags provisioning is likewise unexercised there (RemoteGrpc-only, and that CAP is not listening). #491 stays open for the read-back hop alone. Also recorded, not fixed: the recorder has NO Primary gate, unlike the alarm store-and-forward sink, so both nodes of a redundant pair send duplicate live values. Gating it would introduce a failover window with no capture — that is a design decision, not a cleanup. Evidence: docs/plans/2026-08-01-continuous-historization-live-gate.md Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
+91
-3
@@ -338,6 +338,54 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
|
||||
Assert.Empty(writer.Snapshot());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An append that genuinely YIELDS must not kill the recorder (Gitea #491, live-caught).
|
||||
/// <para>
|
||||
/// The handler awaits <c>IHistorizationOutbox.AppendAsync</c> with <c>ConfigureAwait(false)</c>,
|
||||
/// which discards Akka's ActorTaskScheduler synchronization context. Touching <c>Self</c> after
|
||||
/// that await therefore reads <c>ActorBase.Context</c> with no context installed and throws
|
||||
/// <see cref="NotSupportedException"/> — which escapes into supervision and RESTARTS the actor.
|
||||
/// Production seeds an empty ref set (the deployed set arrives later via
|
||||
/// <see cref="ContinuousHistorizationRecorder.UpdateHistorizedRefs"/>), so a restart silently
|
||||
/// discards every historized ref and historization stops after the FIRST value.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The yield is the whole point: every other test here uses an outbox that returns
|
||||
/// <c>ValueTask.CompletedTask</c>, so the continuation runs inline with the context still
|
||||
/// installed and the defect is invisible. The production FasterLog outbox fsyncs per entry.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_yielding_append_neither_faults_the_actor_nor_loses_the_value()
|
||||
{
|
||||
var mux = CreateTestProbe();
|
||||
var writer = new FakeValueWriter();
|
||||
var outbox = new InMemoryOutbox(yieldOnAppend: true);
|
||||
|
||||
IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
|
||||
mux.Ref, writer, outbox, historizedRefs: new[] { "Pump1.Temp" },
|
||||
drainInterval: TimeSpan.FromMilliseconds(200)));
|
||||
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>();
|
||||
|
||||
var stamp = new DateTime(2026, 8, 1, 12, 0, 0, DateTimeKind.Utc);
|
||||
rec.Tell(new VirtualTagActor.DependencyValueChanged("Pump1.Temp", 42.5, stamp));
|
||||
|
||||
// The value reaches the writer...
|
||||
await AwaitAssertAsync(() =>
|
||||
{
|
||||
Assert.Single(writer.Snapshot());
|
||||
return Task.CompletedTask;
|
||||
});
|
||||
|
||||
// ...and — the load-bearing half — the actor is still the SAME incarnation, holding the ref set
|
||||
// it was seeded with. A restart would reset TotalRecorded to 0 and re-register interest, so this
|
||||
// assertion is what goes red when Self is touched after the await.
|
||||
var status = await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
|
||||
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
|
||||
Assert.Equal(1, status.TotalRecorded);
|
||||
mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // no PreStart re-register == no restart
|
||||
}
|
||||
|
||||
/// <summary>In-memory <see cref="IHistorianValueWriter"/> double: records every value written and
|
||||
/// returns <see cref="Succeed"/> as the ack. Thread-safe — the recorder drains off the actor thread.</summary>
|
||||
private sealed class FakeValueWriter : IHistorianValueWriter
|
||||
@@ -403,7 +451,24 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
|
||||
private readonly int _capacity;
|
||||
private long _dropped;
|
||||
|
||||
public InMemoryOutbox(int capacity = 0) => _capacity = capacity;
|
||||
private readonly bool _yieldOnAppend;
|
||||
|
||||
/// <param name="capacity">Drop-oldest capacity; 0 means unbounded.</param>
|
||||
/// <param name="yieldOnAppend">
|
||||
/// When <c>true</c>, <see cref="AppendAsync"/> genuinely yields before recording the entry —
|
||||
/// modelling the production <c>FasterLogHistorizationOutbox</c>, which fsyncs per entry under
|
||||
/// <c>PerEntry</c> commit mode. This matters because the recorder awaits the append with
|
||||
/// <c>ConfigureAwait(false)</c>: only a real yield resumes the continuation OFF the actor
|
||||
/// thread, where Akka's ActorContext is gone. The default (<c>false</c>) returns
|
||||
/// <c>ValueTask.CompletedTask</c>, so the continuation runs inline with the context still
|
||||
/// installed — which is why every test here passed against a recorder that threw on the very
|
||||
/// first value in production.
|
||||
/// </param>
|
||||
public InMemoryOutbox(int capacity = 0, bool yieldOnAppend = false)
|
||||
{
|
||||
_capacity = capacity;
|
||||
_yieldOnAppend = yieldOnAppend;
|
||||
}
|
||||
|
||||
public long DroppedCount
|
||||
{
|
||||
@@ -417,6 +482,31 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
|
||||
}
|
||||
|
||||
public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct)
|
||||
{
|
||||
if (_yieldOnAppend)
|
||||
{
|
||||
return AppendYieldingAsync(entry);
|
||||
}
|
||||
|
||||
AppendCore(entry);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
private async ValueTask AppendYieldingAsync(HistorizationOutboxEntry entry)
|
||||
{
|
||||
// Must complete on a POOL thread, carrying no synchronization context — that is what a real
|
||||
// per-entry fsync does, and it is what strips the ActorContext from the caller's
|
||||
// ConfigureAwait(false) continuation.
|
||||
//
|
||||
// A bare `await Task.Yield()` does NOT reproduce it: Yield posts back to the CAPTURED context
|
||||
// (Akka's ActorTaskScheduler), so this method completes ON the actor thread, and the caller's
|
||||
// ConfigureAwait(false) then resumes inline right there — with the context still installed.
|
||||
// Verified by reverting the production fix and watching the test still pass.
|
||||
await Task.Delay(1).ConfigureAwait(false);
|
||||
AppendCore(entry);
|
||||
}
|
||||
|
||||
private void AppendCore(HistorizationOutboxEntry entry)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
@@ -427,8 +517,6 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
|
||||
_dropped++;
|
||||
}
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
|
||||
|
||||
Reference in New Issue
Block a user