docs(code): honest rationales — Akka restart semantics, per-key shards, EF literal inlining; retire or annotate orphaned members

This commit is contained in:
Joseph Doherty
2026-08-16 05:33:33 -04:00
parent dbc9223206
commit b9e9df4081
11 changed files with 133 additions and 58 deletions
+11 -4
View File
@@ -281,10 +281,17 @@ non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. T
all 12: nine poll drivers via `Degraded`, MQTT via live connection state, Galaxy via its reconnect all 12: nine poll drivers via `Degraded`, MQTT via live connection state, Galaxy via its reconnect
supervisor. `CalculationDriver` reports `Healthy` while running and so can never trip — correct, as it has supervisor. `CalculationDriver` reports `Healthy` while running and so can never trip — correct, as it has
no transport. Re-check this if a new driver lands. no transport. Re-check this if a new driver lands.
- ⚠️ **The reconnect retry MUST tear down before re-initialising.** `ModbusDriver` (`_transport`) and - **The reconnect retry MUST tear down before re-initialising — FIXED in both drivers, keep it that way.**
`MqttDriver` (`_connection`) overwrite an `IAsyncDisposable` field inside `InitializeAsync` without `ModbusDriver` (`_transport`) and `MqttDriver` (`_connection`) used to overwrite an `IAsyncDisposable`
disposing the old one, so re-initialising without `ShutdownAsync` leaks a socket per attempt — ~360 for an field inside `InitializeAsync` without disposing the old one, so re-initialising without `ShutdownAsync`
hour-long outage at the 10 s retry interval. Latent while only a manual Force Reconnect reached this path. leaked a socket per attempt — ~360 for an hour-long outage at the 10 s retry interval. Both now dispose
first: `ModbusDriver.InitializeCoreAsync` releases `_transport` **and** the `_probeCts`/`_reprobeCts`
probe loops before rebuilding them (`c3970a2e` + the lifecycle gate in `409a58b6`), deliberately *not*
via the shared `TeardownAsync()` because that would also drop `_poll` and every active subscription with
nothing to re-drive them; `MqttDriver.InitializeCoreAsync` calls its (idempotent) `TeardownAsync()` first,
which reaches `_connection` and the primary-host connection (`2b4b3fcb`). The path is no longer latent —
the comms-loss watchdog reaches it, not just a manual Force Reconnect — so a new driver, or a new
`IAsyncDisposable` field on these two, must repeat the dispose-before-overwrite.
- The threshold is hardcoded (`DefaultCommsLossTimeout`), matching `HealthPollInterval` / - The threshold is hardcoded (`DefaultCommsLossTimeout`), matching `HealthPollInterval` /
`DefaultReconnectInterval`. Unlike the #523 drift detector this generates **no** traffic on a healthy `DefaultReconnectInterval`. Unlike the #523 drift detector this generates **no** traffic on a healthy
system — it is a local `GetHealth()` read, and only reaches the network once a driver has already stopped system — it is a local `GetHealth()` read, and only reaches the network once a driver has already stopped
@@ -1413,3 +1413,13 @@ Open items the work surfaced but deliberately did not close. None blocks the bra
`ThreadStatic` actor cell, so under a loaded sweep the intended-red mutation can `ThreadStatic` actor cell, so under a loaded sweep the intended-red mutation can
occasionally pass. The test is still worth keeping; do not read a single green run of it occasionally pass. The test is still worth keeping; do not read a single green run of it
as proof the regression is caught. as proof the regression is caught.
**Accepted-not-fixed at the final integration review (2026-08-15)** — carried knowingly, none
blocking: the poll **pacing** behaviour change ships with no production consumer of the overrun
counter (the per-driver meter follow-up above is the mitigation); **no live gate was run** for this
branch (docker-dev recipe:
[`2026-08-01-continuous-historization-live-gate.md`](2026-08-01-continuous-historization-live-gate.md));
`_globalCount` drift is **negative-only** in direction, so it can only trim early, never late;
`ScriptTagCatalogSnapshotCache` expires on **wall-clock TTL**; the S7 span reader **stamps a partial
span** when the PDU comes back short; MTConnect's `IsFramingOnly` fast path is **ASCII-only**; and a
minor Razor exposure where `StateHasChanged` runs inside a `finally`.
+18 -10
View File
@@ -446,27 +446,35 @@ for the life of a generation, so memoizing the walk is sound. A key's `ExpiresUt
*clock*, not with a deployment — so revocation and expiry are re-checked on every call, outside the *clock*, not with a deployment — so revocation and expiry are re-checked on every call, outside the
cache. Caching them would have let an expired key keep working until the next deploy. cache. Caching them would have let an expired key keep working until the next deploy.
**The cache is sharded per session, because a flat one let any client evict every other client's **The cache is sharded per calling KEY, because a flat one let any client evict every other client's
decisions.** It used to be a single dictionary with the session identity folded *into* the composite key decisions.** It used to be a single dictionary with the caller identity folded *into* the composite key
and one process-wide bound of `MaxEntries = 100_000`, whose only eviction was a wholesale `Clear()` at and one process-wide bound of `MaxEntries = 100_000`, whose only eviction was a wholesale `Clear()` at
the cap. So one machine consumer crawling a large address space past 100k entries wiped the memoized the cap. So one machine consumer crawling a large address space past 100k entries wiped the memoized
walks of every other session on the server, and each store paid a `ConcurrentDictionary.Count` — an walks of every other caller on the server, and each store paid a `ConcurrentDictionary.Count` — an
all-locks snapshot — to test the cap. It is now a dictionary of per-session shards keyed by the session's all-locks snapshot — to test the cap. It is now a dictionary of shards keyed by
`ApiKeyId ?? Principal`, each holding its own `(operation, realm, nodeIdentifier)` entries, with two **`ApiKeyId ?? Principal`**, each holding its own `(operation, realm, nodeIdentifier)` entries, with two
bounds: bounds:
| Bound | Value | What it clears | | Bound | Value | What it clears |
|---|---|---| |---|---|---|
| `MaxEntriesPerSession` | 20 000 | that one session's shard | | `MaxEntriesPerSession` | 20 000 | that one **key's** shard |
| `MaxEntries` | 100 000 | every shard (the shard *objects* are dropped, not emptied) | | `MaxEntries` | 100 000 | every shard (the shard *objects* are dropped, not emptied) |
Both are compile-time constants, not appsettings keys. Five sessions simultaneously at their per-session ⚠️ **The shard key is the API key, not the session — the constant's name predates it.** Every session
ceiling reach the global cap; **the number of shards is not itself bounded** — the bound is entries. authenticating with the same `otua_<keyId>_…` credential shares one shard and one 20 000-entry budget,
so a crawl by one of them *can* clear the memoized decisions of its sibling sessions. What the sharding
guarantees is isolation **across distinct keys**: one key's crawl can never evict another key's entries.
(An Anonymous session shards under the reserved `anonymous` principal, so all of them likewise share
one shard.)
Both bounds are compile-time constants, not appsettings keys. Five keys simultaneously at their
per-key ceiling reach the global cap; **the number of shards is not itself bounded** — the bound is
entries.
Things to know before changing it: Things to know before changing it:
- **There is no session-close eviction hook**, deliberately. A closed session's shard is shed the next - **There is no session-close eviction hook**, deliberately. The shard of a key with no live sessions
time a wholesale clear happens — a generation bump, or the global cap tripping. Dropping the shard left is shed the next time a wholesale clear happens — a generation bump, or the global cap tripping. Dropping the shard
objects rather than emptying them is what makes that sweep work. objects rather than emptying them is what makes that sweep work.
- **The counters are approximate under concurrency, by design.** The cap test is a check-then-act over - **The counters are approximate under concurrency, by design.** The cap test is a check-then-act over
`Volatile.Read` of `Interlocked` counters rather than a lock, so concurrent stores can overshoot a cap `Volatile.Read` of `Interlocked` counters rather than a lock, so concurrent stores can overshoot a cap
@@ -13,7 +13,16 @@ public interface IHistorizationOutbox : IDisposable
/// <summary>Lifetime count of entries dropped because an append would have exceeded capacity.</summary> /// <summary>Lifetime count of entries dropped because an append would have exceeded capacity.</summary>
long DroppedCount { get; } long DroppedCount { get; }
/// <summary>Appends <paramref name="entry"/> to the tail of the durable buffer.</summary> /// <summary>
/// Appends <paramref name="entry"/> to the tail of the durable buffer.
/// <para>
/// <b>Retained for tests and direct callers; production appends only through
/// <see cref="AppendBatchAsync"/>.</b> The recorder coalesces a driver burst into one group
/// commit, so this single-entry form has no production call site — it is kept as the
/// one-entry expression of the same contract (and is what <see cref="AppendBatchAsync"/>'s
/// durability window is described against), not as a deprecated path.
/// </para>
/// </summary>
/// <param name="entry">The value record to buffer.</param> /// <param name="entry">The value record to buffer.</param>
/// <param name="ct">Cancellation token.</param> /// <param name="ct">Cancellation token.</param>
/// <returns>A task that represents the asynchronous operation.</returns> /// <returns>A task that represents the asynchronous operation.</returns>
@@ -5,9 +5,11 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// because a <c>HistorianTagname</c> override makes them diverge: /// because a <c>HistorianTagname</c> override makes them diverge:
/// <list type="bullet"> /// <list type="bullet">
/// <item> /// <item>
/// <see cref="MuxRef"/> — the driver-published reference the per-node dependency mux fans its /// <see cref="MuxRef"/> — the driver-published reference the per-node dependency mux fans values
/// <c>DependencyValueChanged</c> by (the tag's driver-side <c>FullName</c>). The recorder /// by (the tag's driver-side <c>FullName</c>); in production those arrive as a
/// registers mux interest and matches incoming values by THIS. /// <c>DependencyValueBatch</c> per driver burst, with the per-value <c>DependencyValueChanged</c>
/// retained for direct senders and tests. The recorder registers mux interest and matches
/// incoming values by THIS.
/// </item> /// </item>
/// <item> /// <item>
/// <see cref="HistorianName"/> — the resolved historian tag name the value is written under (a /// <see cref="HistorianName"/> — the resolved historian tag name the value is written under (a
@@ -1491,8 +1491,15 @@ public sealed class RawTreeService(IDbContextFactory<OtOpcUaConfigDbContext> dbF
var sourceCode = Expression.Property(parameter, nameof(Script.SourceCode)); var sourceCode = Expression.Property(parameter, nameof(Script.SourceCode));
var contains = typeof(string).GetMethod(nameof(string.Contains), [typeof(string)])!; var contains = typeof(string).GetMethod(nameof(string.Contains), [typeof(string)])!;
// Indexing a captured array (rather than Expression.Constant(path)) keeps EF parameterising each // Indexing an array (rather than Expression.Constant(path) directly) keeps the expression tree
// path instead of inlining it as a SQL literal, so the paths never reach the generated SQL text. // cache-friendly: every rename of the same path-count produces the same tree shape.
// ⚠️ It does NOT guarantee parameterisation. The array here is a constant with no closure over it,
// so EF may well inline each path into the SQL as an escaped literal — which is safe (EF escapes
// the Contains argument either way, so there is no injection surface) but means a rename can emit up
// to ScriptScanBatchSize (200) literals per batch and churn the plan cache. Whether the provider
// actually parameterises this shape is unverified: proving it needs a SQL-Server-backed test, not
// EF InMemory. Recorded in the follow-up register of
// docs/plans/2026-08-15-perf-roadmap-remainder.md.
var values = paths.ToArray(); var values = paths.ToArray();
Expression? body = null; Expression? body = null;
for (var i = 0; i < values.Length; i++) for (var i = 0; i < values.Length; i++)
@@ -316,6 +316,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <summary> /// <summary>
/// Re-authorizes live subscriptions when a deployment installs a new authorization generation. /// Re-authorizes live subscriptions when a deployment installs a new authorization generation.
/// </summary> /// </summary>
/// <remarks>
/// ⚠️ <b>This is the only trigger, and the pass's memo depends on that.</b>
/// <see cref="IsSubscriptionStillPermitted"/> memoizes key revocation and expiry for the span of one
/// pass — safe only while revalidation is deploy-triggered, because a pass is then a momentary
/// event whose verdicts cannot go stale within it. A future periodic trigger must give each tick a
/// fresh memo (it already does — the dictionary is pass-local) and must not be tempted to reuse one
/// across ticks, or an expiry landing between ticks would be served the previous tick's allow.
/// </remarks>
/// <param name="snapshot">The newly published snapshot; unused — the current state is re-read per item.</param> /// <param name="snapshot">The newly published snapshot; unused — the current state is re-read per item.</param>
private void OnAuthorityChanged(DataPlaneAuthoritySnapshot snapshot) => RevalidateMonitoredItems(); private void OnAuthorityChanged(DataPlaneAuthoritySnapshot snapshot) => RevalidateMonitoredItems();
@@ -32,13 +32,16 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
/// deploy costs one repopulation and cannot be wrong. Deployments are rare relative to reads. /// deploy costs one repopulation and cannot be wrong. Deployments are rare relative to reads.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Storage is sharded per session, under a two-level bound.</b> Each session key owns its own /// <b>Storage is sharded per calling KEY, under a two-level bound.</b> The shard key is
/// dictionary, capped at <see cref="MaxEntriesPerSession"/>; overflowing it clears <i>that shard /// <c>ApiKeyId ?? Principal</c>, so <i>sessions sharing one API key share one shard and one
/// only</i>. A second, global <see cref="MaxEntries"/> bound clears every shard, and exists purely /// <see cref="MaxEntriesPerSession"/> cap</i> — the isolation is across distinct keys, not across
/// so many sessions each staying under their own cap cannot add up to unbounded growth. The flat /// sessions. Overflowing a shard clears <i>that shard only</i>, so a crawl by one key cannot evict
/// cache this replaced had only the global bound, which made one machine consumer crawling a large /// another key's entries. A second, global <see cref="MaxEntries"/> bound clears every shard, and
/// tree an eviction event for every other session on the server — the sessions doing ordinary work /// exists purely so many keys each staying under their own cap cannot add up to unbounded growth.
/// paid for the crawl. Both caps are enforced against <see cref="Interlocked"/> counters, never /// The flat cache this replaced had only the global bound, which made one machine consumer
/// crawling a large tree an eviction event for every other caller on the server — the callers
/// doing ordinary work paid for the crawl. Both caps are enforced against
/// <see cref="Interlocked"/> counters, never
/// <see cref="ConcurrentDictionary{TKey,TValue}.Count"/>, which locks the whole table. /// <see cref="ConcurrentDictionary{TKey,TValue}.Count"/>, which locks the whole table.
/// </para> /// </para>
/// </remarks> /// </remarks>
@@ -46,7 +49,7 @@ public sealed class SessionAclDecisionCache
{ {
/// <summary> /// <summary>
/// Process-wide entry ceiling, past which <b>every</b> shard clears. The outer of the two bounds, /// Process-wide entry ceiling, past which <b>every</b> shard clears. The outer of the two bounds,
/// and a rare fallback: it exists so a large fleet of sessions each staying well under /// and a rare fallback: it exists so a large fleet of distinct keys each staying well under
/// <see cref="MaxEntriesPerSession"/> still cannot grow the cache without limit. /// <see cref="MaxEntriesPerSession"/> still cannot grow the cache without limit.
/// </summary> /// </summary>
/// <remarks> /// <remarks>
@@ -59,16 +62,18 @@ public sealed class SessionAclDecisionCache
public const int MaxEntries = 100_000; public const int MaxEntries = 100_000;
/// <summary> /// <summary>
/// Per-session entry ceiling, past which that session's shard alone clears. The inner and normally /// Per-<b>key</b> entry ceiling (the shard key is <c>ApiKeyId ?? Principal</c>, so every session
/// the only bound reached: it is what keeps one machine consumer crawling a 50k-node tree from /// using the same API key shares this one budget), past which that shard alone clears. The inner
/// evicting the decisions of every other session on the server. /// and normally the only bound reached: it is what keeps one machine consumer crawling a 50k-node
/// tree from evicting the decisions taken for every OTHER key on the server. The name predates the
/// shard key and is kept for source compatibility.
/// </summary> /// </summary>
/// <remarks> /// <remarks>
/// Clearing beats evicting <i>within a shard</i>: an LRU needs per-entry bookkeeping on the hot path /// Clearing beats evicting <i>within a shard</i>: an LRU needs per-entry bookkeeping on the hot path
/// to save a trie walk that is itself cheap, so the eviction policy would cost more than the miss it /// to save a trie walk that is itself cheap, so the eviction policy would cost more than the miss it
/// prevents. The failure mode of clearing is a burst of recomputation, which is exactly the steady /// prevents. The failure mode of clearing is a burst of recomputation, which is exactly the steady
/// state of not having a cache at all — and sharding confines that burst to the session that caused /// state of not having a cache at all — and sharding confines that burst to the key that caused it,
/// it, which is the part the old flat cache got wrong. /// which is the part the old flat cache got wrong.
/// </remarks> /// </remarks>
public const int MaxEntriesPerSession = 20_000; public const int MaxEntriesPerSession = 20_000;
@@ -105,7 +110,7 @@ public sealed class SessionAclDecisionCache
/// answer is already memoized, so the allocation the cache exists to avoid is paid on the hit /// answer is already memoized, so the allocation the cache exists to avoid is paid on the hit
/// path too. Read and Browse ask this question once per node, which is precisely where paying /// path too. Read and Browse ask this question once per node, which is precisely where paying
/// for a walk that never runs is worth removing. <see cref="GetOrAdd"/> remains, delegating /// for a walk that never runs is worth removing. <see cref="GetOrAdd"/> remains, delegating
/// here, for callers that genuinely have nothing cheaper to offer. /// here, but has no production caller — it is exercised by the cache's own tests.
/// </para> /// </para>
/// <para> /// <para>
/// The generation is checked (and a stale cache dropped) here rather than in /// The generation is checked (and a stale cache dropped) here rather than in
@@ -187,8 +192,14 @@ public sealed class SessionAclDecisionCache
/// <summary> /// <summary>
/// Returns the memoized decision for this key, computing and storing it when absent or when the /// Returns the memoized decision for this key, computing and storing it when absent or when the
/// generation has moved on. A convenience over <see cref="TryGet"/> + <see cref="Set"/> for callers /// generation has moved on. A convenience over <see cref="TryGet"/> + <see cref="Set"/>.
/// that have no cheaper way to express the walk than a closure. /// <para>
/// <b>Retained for tests; production has no caller.</b> <see cref="DataPlaneAuthorizer"/> uses
/// the split <see cref="TryGet"/>/<see cref="Set"/> pair precisely so a hit pays for no closure,
/// and no other caller has appeared since. Kept because it is the readable single-expression
/// form of the same protocol and is how the cache's tests drive it — not because a production
/// caller is expected back.
/// </para>
/// </summary> /// </summary>
/// <param name="generation">The authorization generation the decision is being taken under.</param> /// <param name="generation">The authorization generation the decision is being taken under.</param>
/// <param name="sessionKey">Stable identity of the calling session — its key id, else its principal.</param> /// <param name="sessionKey">Stable identity of the calling session — its key id, else its principal.</param>
@@ -8,16 +8,20 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
/// <summary> /// <summary>
/// Continuous-historization engine for non-Galaxy (driver) tags. Registers interest with the /// Continuous-historization engine for non-Galaxy (driver) tags. Registers interest with the
/// per-node <see cref="DependencyMuxActor"/> for the configured historized tag refs, then taps the /// per-node <see cref="DependencyMuxActor"/> for the configured historized tag refs, then taps the
/// <see cref="VirtualTagActor.DependencyValueChanged"/> values the mux fans for those refs: /// values the mux fans for those refs — in production a
/// <see cref="VirtualTagActor.DependencyValueBatch"/> per driver burst, with the per-value
/// <see cref="VirtualTagActor.DependencyValueChanged"/> receive retained for direct senders and tests:
/// value → coerce to numeric → append to the durable <see cref="IHistorizationOutbox"/> (the crash /// value → coerce to numeric → append to the durable <see cref="IHistorizationOutbox"/> (the crash
/// boundary) → a self-paced drain writes the outbox through <see cref="IHistorianValueWriter"/>'s /// boundary) → a self-paced drain writes the outbox through <see cref="IHistorianValueWriter"/>'s
/// live-value path and acks (truncates) on success. /// live-value path and acks (truncates) on success.
/// </summary> /// </summary>
/// <remarks> /// <remarks>
/// <para> /// <para>
/// <b>Message path (grounded against the real actors).</b> The mux fans /// <b>Message path (grounded against the real actors).</b> The mux fans each value — carrying
/// <see cref="VirtualTagActor.DependencyValueChanged"/> — which carries only /// only <c>(TagId, Value, TimestampUtc)</c> and <em>no quality</em> — to every registered
/// <c>(TagId, Value, TimestampUtc)</c> and <em>no quality</em> — to every registered subscriber. /// subscriber, batched as a <see cref="VirtualTagActor.DependencyValueBatch"/> on the production
/// path and as a bare <see cref="VirtualTagActor.DependencyValueChanged"/> from a direct sender
/// or a test.
/// The recorder reuses that existing fan-out (no change to <c>DriverHostActor</c>), so values /// The recorder reuses that existing fan-out (no change to <c>DriverHostActor</c>), so values
/// arrive Good-quality by the same convention the scripted-alarm host uses (the mux only /// arrive Good-quality by the same convention the scripted-alarm host uses (the mux only
/// forwards driver-published values); historized samples are recorded with /// forwards driver-published values); historized samples are recorded with
@@ -29,9 +29,11 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
/// </para> /// </para>
/// ///
/// <para> /// <para>
/// <b>Data flow.</b> <see cref="DependencyMuxActor"/> delivers a /// <b>Data flow.</b> <see cref="DependencyMuxActor"/> delivers the values of every tag the
/// <see cref="VirtualTagActor.DependencyValueChanged"/> for every tag the loaded alarms /// loaded alarms depend on — in production as a batched
/// depend on; the host pushes each into the upstream. The engine self-evaluates: its own /// <see cref="VirtualTagActor.DependencyValueBatch"/> per driver burst, with the per-value
/// <see cref="VirtualTagActor.DependencyValueChanged"/> receive retained for direct senders and
/// tests; the host pushes each into the upstream. The engine self-evaluates: its own
/// <see cref="ITagUpstreamSource.SubscribeTag"/> observer re-runs the predicates and raises /// <see cref="ITagUpstreamSource.SubscribeTag"/> observer re-runs the predicates and raises
/// <see cref="ScriptedAlarmEngine.OnEvent"/>. The host does NOT call evaluate — it only /// <see cref="ScriptedAlarmEngine.OnEvent"/>. The host does NOT call evaluate — it only
/// feeds values in and fans emissions out. /// feeds values in and fans emissions out.
@@ -134,9 +136,10 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
/// <summary>Factory method to create Props for a <see cref="ScriptedAlarmHostActor"/>.</summary> /// <summary>Factory method to create Props for a <see cref="ScriptedAlarmHostActor"/>.</summary>
/// <param name="publishActor">The OPC UA publish actor that consumes /// <param name="publishActor">The OPC UA publish actor that consumes
/// <see cref="OpcUaPublishActor.AlarmStateUpdate"/> bridged from engine emissions.</param> /// <see cref="OpcUaPublishActor.AlarmStateUpdate"/> bridged from engine emissions.</param>
/// <param name="mux">Optional dependency multiplexer the host registers interest with so it /// <param name="mux">Optional dependency multiplexer the host registers interest with so it receives
/// receives a <see cref="VirtualTagActor.DependencyValueChanged"/> per dependency tag. Null on the /// its dependency tags' values — a <see cref="VirtualTagActor.DependencyValueBatch"/> per burst in
/// dev/Mac path (no live values).</param> /// production, or a per-value <see cref="VirtualTagActor.DependencyValueChanged"/> from a direct
/// sender or a test. Null on the dev/Mac path (no live values).</param>
/// <param name="upstream">The mux-fed upstream the engine reads + subscribes from. MUST be the /// <param name="upstream">The mux-fed upstream the engine reads + subscribes from. MUST be the
/// same instance the <paramref name="engine"/> was constructed around.</param> /// same instance the <paramref name="engine"/> was constructed around.</param>
/// <param name="engine">The scripted-alarm engine this host owns + disposes.</param> /// <param name="engine">The scripted-alarm engine this host owns + disposes.</param>
@@ -403,10 +406,12 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
/// Loops the per-value handler, preserving in-burst order.</summary> /// Loops the per-value handler, preserving in-burst order.</summary>
/// <remarks> /// <remarks>
/// <b>Mid-batch exception policy: unguarded, deliberately.</b> An exception escaping /// <b>Mid-batch exception policy: unguarded, deliberately.</b> An exception escaping
/// <see cref="OnDependencyChanged"/> drops the remaining items of this burst and restarts the actor /// <see cref="OnDependencyChanged"/> drops the remaining items of this burst and restarts the actor.
/// — exactly what it already did when the same values arrived as N separate messages (the tail is /// That IS wider than the per-message path — Akka restarts do not flush the mailbox, only the
/// then lost to the mailbox flush instead of to the loop), so batching changes the blast radius by /// failing message is dropped, so as N separate messages the same fault would cost one value rather
/// nothing. Catching per item is deliberately NOT done: the handler is a single /// than the burst's tail. Accepted: the engine tolerates a missed intermediate value (the next
/// delivery for that tag re-converges its state). Catching per item is deliberately NOT done: the
/// handler is a single
/// <c>IUpstreamValueSource.Push</c> into an in-memory cache, so a throw means the upstream itself is /// <c>IUpstreamValueSource.Push</c> into an in-memory cache, so a throw means the upstream itself is
/// broken and swallowing it would feed the engine a silently truncated burst. Contrast /// broken and swallowing it would feed the engine a silently truncated burst. Contrast
/// <c>ContinuousHistorizationRecorder.IngestAsync</c>, which DOES catch per item because its /// <c>ContinuousHistorizationRecorder.IngestAsync</c>, which DOES catch per item because its
@@ -8,9 +8,11 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
/// Bridges a driver that implements <see cref="IDependencyConsumer"/> onto the per-node /// Bridges a driver that implements <see cref="IDependencyConsumer"/> onto the per-node
/// <see cref="DependencyMuxActor"/>. Spawned as a child of <c>DriverHostActor</c> for each such /// <see cref="DependencyMuxActor"/>. Spawned as a child of <c>DriverHostActor</c> for each such
/// driver: on start (and on every apply via <see cref="ReRegister"/>) it registers the driver's /// driver: on start (and on every apply via <see cref="ReRegister"/>) it registers the driver's
/// declared <see cref="IDependencyConsumer.DependencyRefs"/> with the mux, and forwards each /// declared <see cref="IDependencyConsumer.DependencyRefs"/> with the mux, and forwards each value
/// <see cref="VirtualTagActor.DependencyValueChanged"/> the mux fans out into /// the mux fans out into <see cref="IDependencyConsumer.OnDependencyValue"/>. The production
/// <see cref="IDependencyConsumer.OnDependencyValue"/>. /// delivery is the batched <see cref="VirtualTagActor.DependencyValueBatch"/> — one message per
/// driver burst; the per-value <see cref="VirtualTagActor.DependencyValueChanged"/> receive remains
/// for direct senders and tests, and the batch handler simply loops it.
/// ///
/// <para> /// <para>
/// This is the exact seam a <c>VirtualTagActor</c> uses to receive its inputs, generalised so a /// This is the exact seam a <c>VirtualTagActor</c> uses to receive its inputs, generalised so a
@@ -76,11 +78,13 @@ public sealed class DependencyConsumerMuxAdapter : ReceiveActor
/// <remarks> /// <remarks>
/// <b>Mid-batch exception policy: unguarded, matching the single-value path.</b> An exception /// <b>Mid-batch exception policy: unguarded, matching the single-value path.</b> An exception
/// escaping <see cref="IDependencyConsumer.OnDependencyValue"/> drops the remaining items of this /// escaping <see cref="IDependencyConsumer.OnDependencyValue"/> drops the remaining items of this
/// burst and restarts the adapter — the same outcome the single-value handler already has (there the /// burst and restarts the adapter. This DOES widen the per-fault loss: Akka restarts do not flush
/// tail is lost to the mailbox flush rather than to the loop), so batching does not widen it. /// the mailbox (only the failing message is dropped), so as N separate messages the same fault
/// Catching per item HERE only would make the batch path quietly more forgiving than the single-value /// would have cost one value, while batched it costs the rest of the burst's tail. Accepted,
/// one for the same consumer fault, so the two are deliberately kept identical: a consumer that needs /// because an <see cref="IDependencyConsumer"/> tolerates missed intermediate values — the next
/// per-value fault tolerance owns it inside its own <c>OnDependencyValue</c>. /// delivery re-converges its state — and catching per item HERE only would make the batch path
/// quietly more forgiving than the single-value one for the same consumer fault. A consumer that
/// needs per-value fault tolerance owns it inside its own <c>OnDependencyValue</c>.
/// </remarks> /// </remarks>
/// <param name="msg">The batch of matched dependency values.</param> /// <param name="msg">The batch of matched dependency values.</param>
private void OnDependencyValueBatch(VirtualTagActor.DependencyValueBatch msg) private void OnDependencyValueBatch(VirtualTagActor.DependencyValueBatch msg)