docs(plans): value-pipeline batching + surgical deploys implementation plans
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,318 @@
|
||||
# Surgical Deploys Implementation Plan
|
||||
|
||||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.
|
||||
|
||||
**Goal:** Make routine config edits non-events instead of fleet-wide outages: a raw-tag attribute edit or a UNS-reference add/remove/re-point no longer tears down the whole address space (killing every client subscription), the artifact is read from SQL once per apply instead of three times, fully JSON-parsed once per actor instead of ~8–11 times, only the delta is materialised instead of the whole composition, blocking apply I/O leaves the hot actor mailboxes, and the artifact blob is deflate-compressed inside the hashed bytes.
|
||||
|
||||
**Architecture:** Three prongs. (1) **Surgical routing:** `AddressSpaceChangeClassifier` learns that a non-alarm `RawTagDelta` and any `UnsReferenceDelta` are surgical-eligible; `AddressSpaceApplier` applies raw-tag attribute edits in place via the existing `ISurgicalAddressSpaceSink.UpdateTagAttributes` (Raw realm) and handles UNS-reference deltas as remove(previous-node)+recreate(current) — the recreate rides the delta materialise pass. Alarm-bearing raw tags keep the full-rebuild path (the WP4 notifier-wiring guard). (2) **Parse/read-once:** Direct mode threads the already-read blob into the rebuild + subscription pushes (the FetchAndCache path is the template, `DriverHostActor.cs:1650-1654`); a new `DeploymentArtifact.Open()` returns a `ParsedDeploymentArtifact` that parses the JSON once and serves all consumers; the materialise passes gain subset overloads so non-rebuild applies materialise only the plan's added/changed nodes. (3) **Mechanics:** the LocalDb cache write leaves the mailbox via `PipeTo`, the Applying-state dispatch deferral stops busy-spinning, and `ConfigComposer` emits a deflate-compressed blob (magic-sniffed on the read side, hash computed over the compressed bytes so `RevisionHash == SHA-256(stored/wire bytes)` stays true).
|
||||
|
||||
**Tech Stack:** C# / .NET 10, System.Text.Json (`JsonDocument`), Akka.NET, System.IO.Compression, xUnit + Shouldly.
|
||||
|
||||
**Branch:** `perf/surgical-deploys` (branched off `perf/value-pipeline-batching` — plans 1 and 2 touch the same actors; do NOT branch off master).
|
||||
|
||||
---
|
||||
|
||||
## Context for implementers (read before any task)
|
||||
|
||||
Verified facts — do not re-derive:
|
||||
|
||||
- **The rebuild cliff:** `AddressSpaceChangeClassifier.Classify` (`src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceChangeClassifier.cs:69-78`) routes ANY non-empty `ChangedRawTags` / `ChangedUnsReferenceVariables` (among others) to `Rebuild`. `RawTagPlan.ReferencingEquipmentPaths` participates in record equality (`AddressSpaceComposer.cs`, `RawTagPlan.Equals`), so adding/removing a `UnsTagReference` flips the backing raw tag to Changed ⇒ full rebuild. The WP4 comment block (`AddressSpaceChangeClassifier.cs:62-68`) is the binding guard: any surgical `ChangedRawTags` path for ALARM tags must re-wire notifiers — we sidestep it by keeping alarm-bearing deltas on the Rebuild path.
|
||||
- **The applier already has every surgical primitive:** `ISurgicalAddressSpaceSink` (`src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/ISurgicalAddressSpaceSink.cs`) — `UpdateTagAttributes(nodeId, writable, historianTagname, dataType, isArray, arrayLength, realm)` (realm-aware, returns false ⇒ caller rebuilds), `RemoveVariableNode`, `RemoveAlarmConditionNode`, `RemoveEquipmentSubtree`, `UpdateFolderDisplayName`. The applier's surgical branch is `AddressSpaceApplier.Apply` (`AddressSpaceApplier.cs:151-224`); its false/throw ⇒ full-rebuild ratchet must be preserved for every new surgical operation.
|
||||
- **Apply flow (publish actor):** `OpcUaPublishActor.HandleRebuild` (`OpcUaPublishActor.cs:361-512`): load blob (msg.Artifact wins; else ConfigDb) → `FeedDataPlaneAuthority(artifact)` (MUST stay ahead of the `plan.IsEmpty` return — #520 stage 4) → `ParseComposition` → `AddressSpacePlanner.Compute(_lastApplied, composition)` → `plan.IsEmpty` return → `_applier.Apply(plan)` → six full-composition materialise passes (`:456-480`) → `AnnounceAddedNodes` when `!outcome.RebuildCalled`. `_lastApplied` advances only after Apply.
|
||||
- **Apply flow (host actor):** Direct mode `ApplyAndAck` (`DriverHostActor.cs:1680-1754`): `ReconcileDrivers(deploymentId)` reads the blob from SQL and RETURNS it (`appliedBlob`), then `RebuildAddressSpace(correlation, deploymentId)` (NO bytes ⇒ publish actor re-reads SQL at `OpcUaPublishActor.cs:578-593`) and `PushDesiredSubscriptions(deploymentId)` (re-reads SQL again). FetchAndCache mode `HandleFetchedForApply` (`:1609-1674`) already does it right: `ReconcileDriversFromBlob(id, bytes)` → `RebuildAddressSpace(corr, id, appliedBlob)` → `PushDesiredSubscriptionsFromArtifact(id, appliedBlob)` → `CacheAppliedArtifact`.
|
||||
- **Parse entry points** (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs`): `ParseDriverInstances(ReadOnlySpan<byte>)` `:89`, `ResolveClusterScope(ReadOnlySpan<byte>, string)` `:513`, `ParseDriverInstances(blob, nodeId)` `:555` (scope+parse), `ParseDataPlaneAuthority(blob, nodeId)` `:591`, `ParseComposition(blob)` `:893`, `ParseComposition(blob, nodeId, onInconsistency)` `:955` (scope+parse+`BuildClusterSets` `:1018`). EVERY one does `JsonDocument.Parse(blob.ToArray())` — a full copy per parse. The private `Build*` helpers already take `JsonElement root`, which is what makes the parse-once refactor mechanical.
|
||||
- **Per-apply parse count today:** host: `ReconcileDrivers*` (scope+instances = 2) + `PushDesiredSubscriptions*` (scope+composition+clustersets = 3) + `CacheAppliedArtifact` (scope = 1); publish actor: `FeedDataPlaneAuthority` (authority+scope = 2) + `ParseComposition` (3). Target: ONE `JsonDocument` per actor per apply.
|
||||
- **Blocking I/O on mailboxes:** `CacheAppliedArtifact` does `.GetAwaiter().GetResult()` (`DriverHostActor.cs:1901-1904`, comment at `:1870-1875` acknowledges it); the Applying-state deferral re-`Forward`s to Self in a hot loop (`DriverHostActor.cs:944-955`); `BeginFetchAndCacheApply` (`:1571-1602`) is the `PipeTo` template.
|
||||
- **Compression constraint (CLAUDE.md, load-bearing):** `RevisionHash == SHA-256(artifact bytes)` — `ConfigComposer` computes the hash of the exact blob, `GrpcDeploymentArtifactFetcher` verifies fetched bytes against it, chunked LocalDb cache stores those bytes. Compressing BEFORE hashing preserves the invariant everywhere (hash of the compressed bytes). The dispatch stays payload-free.
|
||||
- **Invariants to preserve:**
|
||||
1. **#485 "no bytes is no answer"** — a null/empty artifact fails the apply and keeps last-known-good; never let compression or parse-once break that path.
|
||||
2. **`FeedDataPlaneAuthority` runs ahead of `plan.IsEmpty`** — a key-revocation-only artifact must still land.
|
||||
3. **The full-rebuild ratchet** — any surgical false/throw falls back to `SafeRebuild()`.
|
||||
4. **`_lastApplied` advances only on a successful Apply** — the retry diffs against what is actually materialised.
|
||||
5. **Deterministic composition** — `ConfigComposer`'s deterministic sort is what makes revision-hash no-ops real; compression must not reorder anything (it can't — it wraps the final bytes).
|
||||
6. **The classifier stays default-closed** — any delta shape not explicitly whitelisted still lands on `Rebuild`.
|
||||
- **Key records:** `RawTagPlan(TagId, NodeId, ParentNodeId, DriverInstanceId, Name, DataType, Writable, Alarm, ReferencingEquipmentPaths, IsHistorized, HistorianTagname, IsArray, ArrayLength, Realm)` with custom `Equals` (`AddressSpaceComposer.cs:141+`); `UnsReferenceVariable(UnsTagReferenceId, EquipmentId, NodeId, EffectiveName, BackingRawPath, DataType, Writable, Realm)` (`:204+`); `AddressSpacePlan.RawTagDelta(Previous, Current)` / `UnsReferenceDelta(Previous, Current)` (`AddressSpacePlan.cs:134-138`).
|
||||
- **Materialise passes** (`AddressSpaceApplier.cs`): `MaterialiseHierarchy(composition)` `:643`, `MaterialiseRawSubtree(composition)` `:682` (needs `BuildEquipmentIdByFolderPath(composition)` for alarm notifier wiring), `MaterialiseUnsReferences(composition)` `:776` (builds `rawByPath` from `composition.RawTags` for inherited shape/historian), `MaterialiseScriptedAlarms`, `MaterialiseEquipmentTags`, `MaterialiseEquipmentVirtualTags` (in the `:820+` region). All idempotent loops over composition lists.
|
||||
- **Existing tests to keep green:** `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceChangeClassifierTests.cs` + `AddressSpaceChangeClassifierDualNamespaceTests.cs` (some assert `ChangedRawTags ⇒ Rebuild` / `ChangedUnsReferenceVariables ⇒ Rebuild` — those assertions CHANGE deliberately in Task 4; update them to assert the new contract, alarm-delta cases still Rebuild), applier tests, `OpcUaPublishActorApplyFailureTests`, `DriverHostActorFetchAndCacheBootstrapTests`, `tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/DeploymentArtifactProtoTests.cs`, and `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs` (the live gate shape for "subscriptions survive").
|
||||
- **Commit style:** `feat(...)` / `perf(...)` / `refactor(...)` with scope; one commit per task.
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Direct mode threads the already-read blob
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~3 min
|
||||
**Parallelizable with:** Task 2
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs:1726-1730`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/` — extend the existing apply-path test class that exercises `ApplyAndAck` (grep for it; the FetchAndCache tests show the probe idiom)
|
||||
|
||||
**Step 1: Failing test** — drive a Direct-mode apply with a real blob in the fake ConfigDb; assert the publish-actor probe's `RebuildAddressSpace` message carries `Artifact != null` with the exact blob bytes (today it carries null).
|
||||
|
||||
**Step 3: Implement** — in `ApplyAndAck`, replace lines `:1726` and `:1729`:
|
||||
|
||||
```csharp
|
||||
_opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, appliedBlob));
|
||||
PushDesiredSubscriptionsFromArtifact(deploymentId, appliedBlob);
|
||||
```
|
||||
|
||||
(`appliedBlob` is non-null here — the null case returned at `:1706`.) Leave `PushDesiredSubscriptions(deploymentId)` in place for any other caller; if it becomes call-site-dead, delete it and its SQL read. Update the comment block: the publish actor no longer reads central SQL on the Direct path either — `LoadArtifact`/`LoadLatestArtifact` remain only as the legacy/dev fallback for a bytes-less `RebuildAddressSpace`.
|
||||
|
||||
**Step 4:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests` — PASS.
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "perf(runtime): Direct-mode apply threads the in-hand artifact (1 SQL read per apply, not 3)"`
|
||||
|
||||
---
|
||||
|
||||
### Task 2: `DeploymentArtifact.Open()` — parse once
|
||||
|
||||
**Classification:** high-risk (config-parsing seam; behavior must be byte-identical)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 1
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/ParsedDeploymentArtifactTests.cs` (create)
|
||||
|
||||
**Step 1: Failing test** — build a representative artifact JSON (reuse/borrow the builder from existing `DeploymentArtifact` tests — grep `ParseComposition` under tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests for the fixture); assert parity: `using var parsed = DeploymentArtifact.Open(blob);` then `parsed.ResolveClusterScope(node)`, `parsed.ParseDriverInstances(node)`, `parsed.ParseComposition(node, onInconsistency)`, `parsed.ParseDataPlaneAuthority(node)` each equal the static counterparts' results (deep equality on the records; compositions compare with their record equality).
|
||||
|
||||
**Step 3: Implement**
|
||||
- Add:
|
||||
|
||||
```csharp
|
||||
/// <summary>One-parse view over an artifact blob: a single JsonDocument serves every consumer of an
|
||||
/// apply (driver reconcile, subscription push, composition, data-plane authority, cluster scope) instead
|
||||
/// of each entry point re-parsing — at production sizes each static parse was a full LOH copy + document
|
||||
/// build, and one apply performed ~8-11 of them. Dispose when the apply completes.</summary>
|
||||
public sealed class ParsedDeploymentArtifact : IDisposable
|
||||
{
|
||||
private readonly JsonDocument _doc;
|
||||
private readonly byte[] _blob; // decoded (post-decompression) bytes
|
||||
private ClusterScope? _scope; // memoized per node — every consumer resolves the same node
|
||||
private string? _scopeNodeId;
|
||||
internal ParsedDeploymentArtifact(JsonDocument doc, byte[] blob) { _doc = doc; _blob = blob; }
|
||||
public JsonElement Root => _doc.RootElement;
|
||||
public void Dispose() => _doc.Dispose();
|
||||
}
|
||||
```
|
||||
|
||||
- `public static ParsedDeploymentArtifact Open(byte[] blob)` — parse ONCE (`JsonDocument.Parse(blob)` — no `.ToArray()` copy; the caller's array is not mutated afterward on any call path).
|
||||
- Move the body of each static entry point into an instance method on `ParsedDeploymentArtifact` operating on `Root` (the private `Build*` helpers already take `JsonElement root` — pass `Root` straight in). Memoize `ResolveClusterScope(nodeId)` per node id (`_scope`/`_scopeNodeId`) since `ParseDriverInstances`/`ParseComposition`/`ParseDataPlaneAuthority` each begin with the same scope resolution.
|
||||
- Re-express every existing static as a thin wrapper: `using var parsed = Open(blob.ToArray-if-span-else-array); return parsed.X(...)` — public signatures may stay `ReadOnlySpan<byte>` for compatibility (the wrapper pays one copy; the hot path stops using the statics in Task 3). Do not change any output shape.
|
||||
|
||||
**Step 4:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests` + `dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests` — PASS.
|
||||
|
||||
**Step 5:** `git commit -m "refactor(runtime): ParsedDeploymentArtifact — one JsonDocument per apply"`
|
||||
|
||||
---
|
||||
|
||||
### Task 3: Actors consume `Open()`
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** none (needs 1 + 2)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs` (`ReconcileDriversFromBlob`, `PushDesiredSubscriptionsFromArtifact`, `CacheAppliedArtifact` call sites within one apply)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs` (`HandleRebuild` + `FeedDataPlaneAuthority`)
|
||||
|
||||
**Implement (behavior-preserving):**
|
||||
- Host: in `ApplyAndAck` / `HandleFetchedForApply`, `using var parsed = DeploymentArtifact.Open(appliedBlob);` and pass `parsed` down (add internal overloads `ReconcileDriversFromParsed`, `PushDesiredSubscriptionsFromParsed`, `CacheAppliedArtifact(deploymentId, revision, blob, parsed)` — cache still stores the BYTES; it only needs `parsed` for the scope lookup it currently re-parses for). The existing blob-based methods stay as wrappers for the boot paths.
|
||||
- Publish actor: `HandleRebuild` opens once after the #485 guard; `FeedDataPlaneAuthority(parsed, artifactBytes)` (the SHA-256 at `:535` still hashes the raw bytes); the composition parse uses `parsed.ParseComposition(...)`.
|
||||
- Tests: the Task 2 parity tests + the existing apply suites ARE the net; add one assertion-level test only if a seam demands it.
|
||||
|
||||
Run both Runtime + OpcUaServer test projects. Commit: `perf(runtime): apply path parses the artifact once per actor`
|
||||
|
||||
---
|
||||
|
||||
### Task 4: Classifier — surgical eligibility for raw-tag and UNS-reference deltas
|
||||
|
||||
**Classification:** high-risk (the outage-per-edit fix; default-closed semantics)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceChangeClassifier.cs`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/AddressSpaceChangeClassifierTests.cs` + `AddressSpaceChangeClassifierDualNamespaceTests.cs` (update the deliberately-changed assertions; add the new cases)
|
||||
|
||||
**Step 1: Write the failing tests** (these define the new contract):
|
||||
1. `Reference_only_raw_tag_delta_is_not_a_rebuild` — a `RawTagDelta` whose ONLY difference is `ReferencingEquipmentPaths`, `Alarm == null` both sides ⇒ kind is NOT `Rebuild` (with no other deltas ⇒ `AttributeOnly`).
|
||||
2. `Attribute_raw_tag_delta_is_not_a_rebuild` — non-alarm delta differing in `Writable`/`IsHistorized`/`HistorianTagname`/`DataType`/`IsArray`/`ArrayLength` ⇒ not Rebuild.
|
||||
3. `Alarm_bearing_raw_tag_delta_still_rebuilds` — `Alarm != null` on either side ⇒ `Rebuild` (the WP4 guard).
|
||||
4. `Identity_field_raw_tag_delta_still_rebuilds` — e.g. `DriverInstanceId` or `Name` differs ⇒ `Rebuild` (default-closed).
|
||||
5. `Uns_reference_delta_is_not_a_rebuild` — a `UnsReferenceDelta` (re-point or display-name case) ⇒ not Rebuild.
|
||||
6. Combination: UNS-ref ADD + the induced reference-only raw-tag delta ⇒ `PureAdd` (the headline case — one added UNS variable, zero teardown).
|
||||
|
||||
**Step 3: Implement**
|
||||
- Add the predicate (mirror `TagDeltaIsSurgicalEligible`'s whitelist-via-`with` idiom, which auto-closes over future fields):
|
||||
|
||||
```csharp
|
||||
// v3 surgical raw-tag eligibility: a NON-alarm raw tag whose only differences are the in-place-updatable
|
||||
// attributes (Writable / IsHistorized / HistorianTagname / DataType / IsArray / ArrayLength) and/or its
|
||||
// ReferencingEquipmentPaths. The referencing set is node-inert for a VALUE tag — it only drives native-alarm
|
||||
// notifier wiring (the WP4 guard), which is why Alarm != null on EITHER side disqualifies. Identity fields
|
||||
// (NodeId is the diff key; TagId / DriverInstanceId / Name / ParentNodeId / Realm) still force a rebuild via
|
||||
// the with-equality default-closed pattern.
|
||||
internal static bool RawTagDeltaIsSurgicalEligible(AddressSpacePlan.RawTagDelta d) =>
|
||||
d.Previous.Alarm is null && d.Current.Alarm is null &&
|
||||
(d.Previous with
|
||||
{
|
||||
Writable = d.Current.Writable,
|
||||
IsHistorized = d.Current.IsHistorized,
|
||||
HistorianTagname = d.Current.HistorianTagname,
|
||||
DataType = d.Current.DataType,
|
||||
IsArray = d.Current.IsArray,
|
||||
ArrayLength = d.Current.ArrayLength,
|
||||
ReferencingEquipmentPaths = d.Current.ReferencingEquipmentPaths,
|
||||
}).Equals(d.Current);
|
||||
```
|
||||
|
||||
- In `Classify` rule 2 (`:69-78`): replace `plan.ChangedRawTags.Count > 0` with `plan.ChangedRawTags.Any(d => !RawTagDeltaIsSurgicalEligible(d))`, and DELETE `plan.ChangedUnsReferenceVariables.Count > 0` from the rebuild condition (every `UnsReferenceDelta` is handled surgically in Tasks 5-6: remove-previous + recreate-current).
|
||||
- Rewrite the WP4 comment block to state the NEW contract: alarm-bearing raw-tag deltas still rebuild (notifier wiring); value-tag deltas are surgical; the applier owns `UpdateTagAttributes` for them; UNS-ref deltas are remove+recreate.
|
||||
|
||||
**Step 4:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests --filter "FullyQualifiedName~AddressSpaceChangeClassifier"` — PASS (including the updated legacy assertions).
|
||||
|
||||
**Step 5:** `git commit -m "feat(opcuaserver): classifier — non-alarm raw-tag + UNS-reference deltas are surgical, not rebuilds"`
|
||||
|
||||
---
|
||||
|
||||
### Task 5: Applier — apply the new surgical deltas
|
||||
|
||||
**Classification:** high-risk (rebuild ratchet semantics)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (needs 4)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs` (`Apply`, `:151-224`)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/` — extend the applier test class (find it via grep `AddressSpaceApplier` in that project; use its recording-fake surgical sink)
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `Surgical_raw_tag_delta_updates_in_place_raw_realm` — plan with one eligible `RawTagDelta` (historize flip); fake surgical sink records `UpdateTagAttributes(d.Current.NodeId, writable, historian, dataType, isArray, arrayLength, AddressSpaceRealm.Raw)` with historian resolved as `MaterialiseRawSubtree` resolves it (override else NodeId, `AddressSpaceApplier.cs:748-750`) and `writable = Writable && !IsArray`; NO `RebuildAddressSpace` call.
|
||||
2. `Reference_only_raw_tag_delta_is_a_noop_update` — only `ReferencingEquipmentPaths` differs ⇒ either no sink call or one idempotent `UpdateTagAttributes` (assert no rebuild; prefer SKIPPING the call when the with-comparison excluding the referencing set shows attribute equality — implement that skip).
|
||||
3. `Uns_reference_delta_removes_previous_node` — plan with one `UnsReferenceDelta`; sink records terminal-Bad `WriteValue(Previous.NodeId, ..., Uns)` then `RemoveVariableNode(Previous.NodeId, Uns)`; no rebuild. (The recreate happens in the publish actor's delta materialise pass — Task 6 — NOT here.)
|
||||
4. `Surgical_false_falls_back_to_rebuild` — surgical sink returns false for the raw-tag update ⇒ `RebuildAddressSpace` called (ratchet preserved).
|
||||
|
||||
**Step 3: Implement** in `Apply`:
|
||||
- Compute `var surgicalRawTagDeltas = plan.ChangedRawTags.Where(AddressSpaceChangeClassifier.RawTagDeltaIsSurgicalEligible).ToList();` (when kind != Rebuild this equals ALL of ChangedRawTags — the classifier guarantees it; still filter, defense-in-depth) and `var unsRefDeltas = plan.ChangedUnsReferenceVariables;`
|
||||
- Extend the surgical-branch trigger (`:171`): `surgicalTagDeltas.Count > 0 || renamedFolders.Count > 0 || hasRemovePass || surgicalRawTagDeltas.Count > 0 || unsRefDeltas.Count > 0`.
|
||||
- Inside the `ISurgicalAddressSpaceSink` branch, after the existing equipment-tag loop:
|
||||
- Raw-tag loop: skip when attribute-equal apart from `ReferencingEquipmentPaths` (test 2); else `surgical.UpdateTagAttributes(d.Current.NodeId, d.Current.Writable && !d.Current.IsArray, d.Current.IsHistorized ? (string.IsNullOrWhiteSpace(d.Current.HistorianTagname) ? d.Current.NodeId : d.Current.HistorianTagname) : null, d.Current.DataType, d.Current.IsArray, d.Current.ArrayLength, AddressSpaceRealm.Raw)` with the same try/catch + `allApplied` ratchet as the equipment-tag loop.
|
||||
- UNS-ref loop: `SafeWriteValue(d.Previous.NodeId, null, OpcUaStatus.Bad, ts, AddressSpaceRealm.Uns);` then `SafeRemoveVariable(surgical, d.Previous.NodeId, AddressSpaceRealm.Uns)` into the ratchet. Doc-comment WHY remove+recreate (a re-point must move the `Organizes` edge and the sink has no RemoveReference; a display-name edit changes the NodeId; remove+recreate covers both, at the cost of that one node's monitored items — vs. every subscription server-wide today).
|
||||
- `changedCount` accounting already includes both sets (`:115-121`) — verify, don't double-count.
|
||||
- Also update the `FeedHistorizedRefs` changed-raw-tags handling: it already handles `plan.ChangedRawTags` (`:571-578`) — now those deltas reach it on the surgical path too; confirm the logic is delta-shape-agnostic (it is) and leave it.
|
||||
|
||||
**Step 4:** run the OpcUaServer.Tests project — PASS.
|
||||
|
||||
**Step 5:** `git commit -m "feat(opcuaserver): applier applies raw-tag + UNS-reference deltas surgically"`
|
||||
|
||||
---
|
||||
|
||||
### Task 6: Delta materialisation (publish actor materialises the plan, not the world)
|
||||
|
||||
**Classification:** high-risk (the O(fleet)-per-apply fix)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (needs 5)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs` (subset overloads on the six passes)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs:456-480`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/` applier tests + `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/` publish-actor tests
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `Pure_add_materialises_only_the_added_nodes` — publish-actor-level: apply composition A (N nodes), then A+1 tag; the fake sink's `EnsureVariable`/`EnsureFolder` calls after the second apply cover ONLY the added tag (+ its added containers), not all N. (This is the headline test.)
|
||||
2. `Rebuild_still_materialises_the_full_composition` — force a rebuild-kind delta; full-composition passes run as today.
|
||||
3. `Uns_reference_delta_recreates_current_node` — a re-point delta: after Task 5's remove, the UNS pass re-ensures `Current.NodeId` with shape inherited from the FULL composition's `rawByPath` (backing tag NOT in the delta), plus the `AddReference` Organizes edge to the NEW backing path.
|
||||
4. Applier-level: `MaterialiseRawSubtree(full, subsetTags, subsetContainers)` wires alarm notifiers using the FULL composition's equipment map.
|
||||
|
||||
**Step 3: Implement**
|
||||
- Applier: give the passes subset parameters, full-composition context retained for lookups:
|
||||
- `MaterialiseRawSubtree(AddressSpaceComposition composition, IReadOnlyList<RawContainerNode>? onlyContainers = null, IReadOnlyList<RawTagPlan>? onlyTags = null)` — iterate `onlyContainers ?? composition.RawContainers` / `onlyTags ?? composition.RawTags`; `BuildEquipmentIdByFolderPath(composition)` stays full.
|
||||
- `MaterialiseUnsReferences(AddressSpaceComposition composition, IReadOnlyList<UnsReferenceVariable>? only = null)` — iterate `only ?? composition.UnsReferenceVariables`; `rawByPath` stays full.
|
||||
- `MaterialiseScriptedAlarms` / `MaterialiseEquipmentTags` / `MaterialiseEquipmentVirtualTags`: same `only` pattern over their respective lists.
|
||||
- `MaterialiseHierarchy` stays full-composition ALWAYS (folders are O(areas+lines+equipment), cheap, and parents must exist for any add).
|
||||
- Publish actor (`:456-480`): after `var outcome = _applier.Apply(plan);` branch on `outcome.RebuildCalled`:
|
||||
- Rebuild ⇒ the six passes exactly as today (full composition).
|
||||
- Non-rebuild ⇒ `MaterialiseHierarchy(composition)` full; then subsets: containers = `plan.AddedRawContainers`, raw tags = `plan.AddedRawTags`, UNS refs = `plan.AddedUnsReferenceVariables` concat `plan.ChangedUnsReferenceVariables.Select(d => d.Current)`, scripted alarms = `plan.AddedAlarms`, equipment tags = `plan.AddedEquipmentTags`, vtags = `plan.AddedEquipmentVirtualTags`. Skip a pass entirely when its subset is empty.
|
||||
- Comment WHY: the passes are idempotent but not free — each call is a lock acquisition; a one-tag add on a 50k-node server was ~50k sink calls.
|
||||
- `AnnounceAddedNodes` stays as-is (known pre-existing gap: raw/UNS adds are not announced — out of scope, note it in the comment).
|
||||
|
||||
**Step 4:** OpcUaServer.Tests + Runtime.Tests — PASS (watch `OpcUaPublishActorApplyFailureTests` + `DarkAddressSpaceReasons` for full-pass assumptions; update only assertions that pinned the OLD full-pass behavior explicitly, preserving their intent).
|
||||
|
||||
**Step 5:** `git commit -m "perf(opcuaserver): non-rebuild applies materialise the delta, not the composition"`
|
||||
|
||||
---
|
||||
|
||||
### Task 7: Blocking I/O off the host mailbox
|
||||
|
||||
**Classification:** high-risk (actor lifecycle)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 8
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs` (`CacheAppliedArtifact` `:1896-1904` region; Applying-state deferral `:944-955`)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/` — extend the apply/dispatch test classes
|
||||
|
||||
**Part A — async cache write.** `CacheAppliedArtifact`'s `.GetAwaiter().GetResult()` becomes fire-and-forget with observed completion:
|
||||
|
||||
```csharp
|
||||
_artifactCache.StoreAsync(...).PipeTo(Self,
|
||||
success: () => new ArtifactCached(deploymentId),
|
||||
failure: ex => new ArtifactCacheFailed(deploymentId, ex.Message));
|
||||
```
|
||||
|
||||
(private records; handlers just log Info/Warning — the cache is a boot fallback, not apply-critical; the docstring at `:1870-1875` already debates exactly this and the apply outcome does not depend on it). CHECK first whether `BootstrapFromCache` (`:788, :859`) races a concurrent store — it runs in PreStart before any apply, so no; note that in the comment. Keep the store idempotent-skip (`IsAlreadyCachedAsync`) inside the async flow.
|
||||
Failing test: apply completes and acks BEFORE a deliberately-slow fake cache resolves; a follow-up `ArtifactCached` self-message is observed (TestKit probe or log assertion per the project's idiom).
|
||||
|
||||
**Part B — stash instead of forward-spin.** The Applying-state deferral for a DIFFERENT deployment (`:944-955`): make `DriverHostActor` implement `IWithUnboundedStash` (`public IStash Stash { get; set; }`), replace `Self.Forward(msg)` with `Stash.Stash()`, and call `Stash.UnstashAll()` at every `Become(Steady)` transition (both `finally` blocks + the fetch-failure path). Failing test: send dispatch B while A's fetch is in flight (slow fake fetcher); assert no busy-loop (the log does NOT contain the deferral line more than once) and B applies after A completes.
|
||||
|
||||
**Step 4:** Runtime.Tests — PASS. **Step 5:** `git commit -m "perf(runtime): apply-path cache write off-mailbox; stash deferred dispatches"`
|
||||
|
||||
---
|
||||
|
||||
### Task 8: Compressed artifact blob
|
||||
|
||||
**Classification:** high-risk (wire/storage format; hash invariant)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 7
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/ConfigComposer.cs` (emit compressed; hash unchanged in derivation — SHA-256 of the emitted bytes)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs` (`Open()` + static wrappers sniff-and-decompress)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/ParsedDeploymentArtifactTests.cs` + the ConfigComposer test class (grep `SnapshotAndFlattenAsync` under tests/Server)
|
||||
|
||||
**Format decision (locked):** gzip via `System.IO.Compression.GZipStream` (`CompressionLevel.Fastest`), detected by the gzip magic bytes `0x1F 0x8B` at offset 0 (a JSON artifact can never start with 0x1F). No custom framing. Compression happens INSIDE `ConfigComposer` immediately after `SerializeToUtf8Bytes`, BEFORE the revision hash is computed — so `RevisionHash == SHA-256(stored bytes)` holds and `GrpcDeploymentArtifactFetcher`'s verification (`GrpcDeploymentArtifactFetcher.cs:89-90`) is untouched. The LocalDb cache + gRPC serve stores/ships the compressed bytes as-is (they never look inside). Decompression lives in exactly one place: `DeploymentArtifact.Open()` (and thus every static wrapper). `OpcUaPublishActor.FeedDataPlaneAuthority`'s SHA at `OpcUaPublishActor.cs:535` hashes the RAW (compressed) bytes it received — verify that call site hashes the bytes as-carried, not a decompressed copy.
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `Open_reads_gzip_compressed_artifacts` — compress a fixture blob, parity-assert `ParseComposition` vs the uncompressed parse.
|
||||
2. `Open_still_reads_uncompressed_artifacts` (backward compat — pre-upgrade rows).
|
||||
3. `Composer_emits_gzip_and_hash_of_compressed_bytes` — `ArtifactBlob` starts with `1F 8B`; the deployment's `RevisionHash` equals `SHA256(ArtifactBlob)` lowercase hex.
|
||||
|
||||
**Step 3: Implement** — decompress with a size guard (refuse decompressed output > 512 MB — a corrupt/hostile blob must not OOM the node; treat over-limit exactly like unparseable JSON: the existing catch paths). Note in a comment: gzip output is not guaranteed byte-stable across .NET releases; the hash is computed from the emitted bytes at compose time and never re-derived, so the only consequence of an impl change is one spurious "changed" revision after a runtime upgrade.
|
||||
|
||||
**Step 4:** Runtime.Tests + ControlPlane tests (if the composer test project is DB-backed and the DB is unreachable, run the artifact round-trip tests and note the skip). **Step 5:** `git commit -m "perf(controlplane): gzip artifact blob inside the hashed bytes (~10-20x smaller)"`
|
||||
|
||||
---
|
||||
|
||||
### Task 9: Full verification + docs
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (last)
|
||||
|
||||
**Files:**
|
||||
- Modify: `CLAUDE.md` — the "v3 OPC UA Address Space" / classifier-adjacent statements that now describe the old behavior (grep CLAUDE.md for `ChangedRawTags` and `Rebuild`); keep the edits surgical and truthful to the new contract (alarm-bearing deltas still rebuild).
|
||||
- Modify: `docs/Uns.md` + `docs/Raw.md` — one short paragraph each: reference add/remove and attribute edits are now surgical; what still rebuilds (alarm deltas, container renames, identity changes).
|
||||
|
||||
Steps: full `dotnet build ZB.MOM.WW.OtOpcUa.slnx`; run every non-DB test project (same list as plan 1 Task 12 + `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests`); record skipped DB-backed suites in the commit message. Commit: `docs: record surgical-deploy semantics (plan 2026-08-15)`.
|
||||
|
||||
---
|
||||
|
||||
## Explicit non-goals
|
||||
|
||||
- **No surgical raw-container (folder/device/group) renames** — still rebuild (no surgical folder-remove on the sink; pre-existing).
|
||||
- **No alarm-bearing raw-tag surgical path** — the WP4 notifier-rewire stays owned by the rebuild.
|
||||
- **No central-side deploy-start dedup** (`AdminOperationsActor` triple DB read) — separate follow-up.
|
||||
- **No `AnnounceAddedNodes` extension to raw/UNS adds** — pre-existing gap, noted in comments.
|
||||
- **No cross-actor shared `ParsedDeploymentArtifact`** — one parse per actor is the target; sharing a disposable across mailboxes isn't worth the ownership complexity.
|
||||
- **No artifact schema changes** — the JSON inside the (now compressed) blob is byte-identical.
|
||||
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"planPath": "docs/plans/2026-08-15-surgical-deploys.md",
|
||||
"tasks": [
|
||||
{"id": 1, "subject": "Task 1: Direct mode threads the already-read blob", "status": "pending"},
|
||||
{"id": 2, "subject": "Task 2: DeploymentArtifact.Open() — parse once", "status": "pending"},
|
||||
{"id": 3, "subject": "Task 3: Actors consume Open()", "status": "pending", "blockedBy": [1, 2]},
|
||||
{"id": 4, "subject": "Task 4: Classifier — surgical eligibility for raw-tag and UNS-reference deltas", "status": "pending"},
|
||||
{"id": 5, "subject": "Task 5: Applier — apply the new surgical deltas", "status": "pending", "blockedBy": [4]},
|
||||
{"id": 6, "subject": "Task 6: Delta materialisation", "status": "pending", "blockedBy": [5]},
|
||||
{"id": 7, "subject": "Task 7: Blocking I/O off the host mailbox", "status": "pending", "blockedBy": [3]},
|
||||
{"id": 8, "subject": "Task 8: Compressed artifact blob", "status": "pending", "blockedBy": [2]},
|
||||
{"id": 9, "subject": "Task 9: Full verification + docs", "status": "pending", "blockedBy": [3, 6, 7, 8]}
|
||||
],
|
||||
"lastUpdated": "2026-08-15T00:00:00Z"
|
||||
}
|
||||
@@ -0,0 +1,579 @@
|
||||
# Value-Pipeline Batching Implementation Plan
|
||||
|
||||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers-extended-cc:executing-plans to implement this plan task-by-task.
|
||||
|
||||
**Goal:** Batch the hot value path end to end — one message per poll cycle instead of one per tag at every hop (driver → DriverInstanceActor → DriverHostActor → OpcUaPublishActor → node manager), one SDK-lock acquisition per batch instead of per value, and zero per-write key allocations in the node manager.
|
||||
|
||||
**Architecture:** A new optional `IBatchSubscribable` capability lets a driver raise one `OnDataChangeBatch` per production burst (poll cycle); `PollGroupEngine` gains a batch callback so the four engine-based poll drivers get it for free. `DriverInstanceActor` forwards one `AttributeValueBatchPublished` per burst (wrapping per-tag events from non-batch drivers as single-item batches); `DriverHostActor` resolves the whole batch's NodeRealmRefs and sends ONE `AttributeValueBatchUpdate` to `OpcUaPublishActor`, plus ONE batch Tell to the dependency mux. The sink gains `WriteValues(IReadOnlyList<ValueWrite>)` (default-interface-method fallback loops `WriteValue`, so fakes/Null sink are untouched); `OtOpcUaNodeManager` overrides it taking `Lock` once per batch, and its node maps are re-keyed by a `(realm, id)` struct so the per-write `new NodeId(...).ToString()` allocations disappear. The pinned dispatcher's `throughput` rises from 1 to 100 (same single thread, fewer mailbox re-schedules).
|
||||
|
||||
**Tech Stack:** C# / .NET 10, Akka.NET (classic actors), OPC Foundation UA .NET Standard, xUnit + Shouldly, Akka.TestKit.
|
||||
|
||||
**Branch:** `perf/value-pipeline-batching` (off `master`).
|
||||
|
||||
---
|
||||
|
||||
## Context for implementers (read before any task)
|
||||
|
||||
Verified facts about the current code — do not re-derive, and do not violate the invariants:
|
||||
|
||||
- **The per-tag path today:** `PollGroupEngine.PollOnceAsync` (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs:202-226`) diffs per tag and invokes `_onChange` per changed tag. Drivers forward that to `ISubscribable.OnDataChange` (per-tag event, `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs:36`). `DriverInstanceActor.HandleSubscribeAsync` attaches `_dataChangeHandler` at `DriverInstanceActor.cs:880` (one `DataChangeForward` Tell per tag), `OnDataChangeForward` (`:1138-1143`) Tells parent one `AttributeValuePublished` per tag. `DriverHostActor.ForwardToMux` (`DriverHostActor.cs:976-1006`) Tells the mux per value and Tells `OpcUaPublishActor` one `AttributeValueUpdate` **per NodeRealmRef** (UNS fan-out at `:997-999`). `OpcUaPublishActor.HandleAttributeUpdate` (`OpcUaPublishActor.cs:319-331`) calls `_sink.WriteValue` per message. `OtOpcUaNodeManager.WriteValue` (`OtOpcUaNodeManager.cs:508-533`) takes the manager-wide `lock (Lock)` per value and allocates via `MapKey(realm, id)` = `new NodeId(id, ns).ToString()` (`:133`).
|
||||
- **Invariants to preserve:**
|
||||
1. **#519 status codes ride verbatim** — the full `uint` status code, sub-code included, must reach the SDK unchanged at every hop.
|
||||
2. **Timestamps are carried from the driver** — never stamp `DateTime.UtcNow` on the value path.
|
||||
3. **PollGroupEngine's no-overlap guarantee** (delay-after-completion, one loop task per subscription) must not change in this plan.
|
||||
4. **Per-tag `OnDataChange` keeps working** — Galaxy's cold-read path and driver CLIs subscribe it. A batch-capable driver raises BOTH events (batch once per cycle + per-tag per item); a consumer subscribes to exactly ONE of the two. `DriverInstanceActor` prefers batch when the driver implements `IBatchSubscribable`.
|
||||
5. **`DriverInstanceActor.DetachSubscription`** (`:934-943`) must detach the batch handler too — the detach is load-bearing (the `IDriver` outlives the actor across respawns; see CLAUDE.md §DeployWatcher).
|
||||
6. **The mux's per-item receive stays** — `DriverHostActor.FanCommsLossToDependencyMux` (`:1135-1142`) fabricates per-item `AttributeValuePublished` messages; `DependencyMuxActor` keeps that receive and gains a batch receive alongside.
|
||||
7. **`OpcUaPublishActor` keeps its per-item `AttributeValueUpdate` receive** — `VirtualTagHostActor` and the alarm paths still send it. The batch message is additive.
|
||||
8. **The pinned dispatcher stays pinned** — the OPC UA SDK requires a single thread. Only `throughput` changes.
|
||||
9. **`_writes` increments** (`OpcUaPublishActor.cs:324`) and the `OtOpcUaTelemetry.OpcUaSinkWrite` meter must count per-VALUE (per item), not per batch, so dashboards don't drop by 100×.
|
||||
- **Message-record homes:** `AttributeValuePublished` is declared in `DriverInstanceActor` (`:109`); `AttributeValueUpdate` in `OpcUaPublishActor` (`:52`); `NodeRealmRef` is a private struct of `DriverHostActor` (`:183`).
|
||||
- **Sink implementors that must keep compiling:** `NullOpcUaAddressSpaceSink` (same file as the interface), `DeferredAddressSpaceSink` (`src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs`), `SdkAddressSpaceSink` (`src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs`), plus test fakes. Using a **default interface method** for `WriteValues` means only the ones that want batching override it.
|
||||
- **Commands:** build `dotnet build ZB.MOM.WW.OtOpcUa.slnx`; single project `dotnet test tests/<module>/<project>`; single test `dotnet test --filter "FullyQualifiedName~<Class>.<Method>"`. DB-backed suites need the central SQL Server — on this Mac, prefer running the specific unit projects named per task, not the whole solution, until the final task.
|
||||
- **Commit style:** follow recent history (`feat(...)`, `fix(...)`, `chore(...)` with a scope). One commit per task.
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Batch contract types in Core.Abstractions
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~3 min
|
||||
**Parallelizable with:** Task 10
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs`
|
||||
- Test: `tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/DataChangeBatchContractTests.cs` (create)
|
||||
|
||||
**Step 1: Write the failing test**
|
||||
|
||||
```csharp
|
||||
using Shouldly;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.Tests;
|
||||
|
||||
public class DataChangeBatchContractTests
|
||||
{
|
||||
private sealed record FakeHandle(string DiagnosticId) : ISubscriptionHandle;
|
||||
|
||||
[Fact]
|
||||
public void Batch_args_carry_handle_and_items()
|
||||
{
|
||||
var handle = new FakeHandle("h1");
|
||||
var snap = new DataValueSnapshot(42, 0u, DateTime.UtcNow, DateTime.UtcNow);
|
||||
var args = new DataChangeBatchEventArgs(handle, new[] { new DataChangeItem("plc/tag1", snap) });
|
||||
args.SubscriptionHandle.ShouldBe(handle);
|
||||
args.Items.Count.ShouldBe(1);
|
||||
args.Items[0].FullReference.ShouldBe("plc/tag1");
|
||||
args.Items[0].Snapshot.ShouldBe(snap);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Note: check `DataValueSnapshot`'s actual constructor shape in `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/` first and adjust the test's construction to it (positional record with Value/StatusCode/timestamps). Do not change `DataValueSnapshot`.
|
||||
|
||||
**Step 2: Run test to verify it fails**
|
||||
|
||||
Run: `dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests --filter "FullyQualifiedName~DataChangeBatchContractTests"`
|
||||
Expected: FAIL — `DataChangeBatchEventArgs` / `DataChangeItem` not defined.
|
||||
|
||||
**Step 3: Implement**
|
||||
|
||||
Append to `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs`:
|
||||
|
||||
```csharp
|
||||
/// <summary>One changed attribute within a <see cref="DataChangeBatchEventArgs"/>.</summary>
|
||||
/// <param name="FullReference">Driver-side full reference of the changed attribute.</param>
|
||||
/// <param name="Snapshot">New value + quality + timestamps.</param>
|
||||
public sealed record DataChangeItem(string FullReference, DataValueSnapshot Snapshot);
|
||||
|
||||
/// <summary>Event payload for <see cref="IBatchSubscribable.OnDataChangeBatch"/> — every attribute
|
||||
/// that changed within ONE production burst (a poll cycle, an inbound message, a stream chunk).</summary>
|
||||
/// <param name="SubscriptionHandle">The handle returned by the original <see cref="ISubscribable.SubscribeAsync"/> call.</param>
|
||||
/// <param name="Items">The changed attributes, one entry per changed tag. Never empty.</param>
|
||||
public sealed record DataChangeBatchEventArgs(
|
||||
ISubscriptionHandle SubscriptionHandle,
|
||||
IReadOnlyList<DataChangeItem> Items);
|
||||
|
||||
/// <summary>
|
||||
/// Optional driver capability: data changes delivered one event per production burst instead of
|
||||
/// one event per tag. A driver implementing this raises <see cref="OnDataChangeBatch"/> once per
|
||||
/// burst AND continues raising the per-tag <see cref="ISubscribable.OnDataChange"/> for each item
|
||||
/// (legacy consumers: Galaxy cold reads, driver CLIs). A consumer subscribes to EXACTLY ONE of the
|
||||
/// two events — subscribing both double-delivers. DriverInstanceActor prefers the batch event.
|
||||
/// </summary>
|
||||
public interface IBatchSubscribable : ISubscribable
|
||||
{
|
||||
/// <summary>Batched data-change notification — one raise per production burst.</summary>
|
||||
event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
|
||||
}
|
||||
```
|
||||
|
||||
**Step 4: Run test to verify it passes**
|
||||
|
||||
Run: same filter. Expected: PASS.
|
||||
|
||||
**Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/ISubscribable.cs tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/DataChangeBatchContractTests.cs
|
||||
git commit -m "feat(core): add IBatchSubscribable batch data-change contract"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: PollGroupEngine batch callback
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (Task 1 first)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs`
|
||||
- Test: `tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/PollGroupEngineBatchTests.cs` (create)
|
||||
|
||||
**Step 1: Write the failing tests** (find the existing PollGroupEngine test file first — if one exists in `tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests`, put these in a new class beside it and copy its reader-fake idiom):
|
||||
|
||||
Test cases (write all four):
|
||||
1. `Batch_fires_once_per_cycle_with_only_changed_items` — reader returns 3 refs; first poll (initial push) batch has all 3; then reader changes only 1 value; next batch has exactly that 1 item; per-tag `onChange` also fired for the same items (both callbacks see identical content).
|
||||
2. `No_batch_when_nothing_changed` — second poll with identical snapshots ⇒ no batch invocation (and no per-tag invocation).
|
||||
3. `Batch_is_not_invoked_when_callback_absent` — construct without `onBatch`; engine behaves exactly as today (per-tag only). Assert per-tag count.
|
||||
4. `Batch_items_are_in_reader_order`.
|
||||
|
||||
Use a short interval (e.g. 30 ms) + `TaskCompletionSource`/polling-with-timeout to await callbacks, mirroring the existing engine tests' style. Never `Thread.Sleep` long constants.
|
||||
|
||||
**Step 2: Run to verify failure**
|
||||
|
||||
Run: `dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests --filter "FullyQualifiedName~PollGroupEngineBatchTests"`
|
||||
Expected: FAIL — constructor has no `onBatch` parameter.
|
||||
|
||||
**Step 3: Implement**
|
||||
|
||||
In `PollGroupEngine`:
|
||||
- Add field `private readonly Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? _onBatch;`
|
||||
- Add optional ctor param `Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? onBatch = null` (AFTER the existing optional params, so every existing call site compiles unchanged) and assign it.
|
||||
- In `PollOnceAsync` (currently `:202-226`): collect changed items lazily and raise the batch after the loop:
|
||||
|
||||
```csharp
|
||||
List<DataChangeItem>? changed = null;
|
||||
for (var i = 0; i < state.TagReferences.Count; i++)
|
||||
{
|
||||
var tagRef = state.TagReferences[i];
|
||||
var current = snapshots[i];
|
||||
var lastSeen = state.LastValues.TryGetValue(tagRef, out var prev) ? prev : default;
|
||||
|
||||
if (forceRaise || ValuesAreDifferent(lastSeen?.Value, current.Value) || lastSeen?.StatusCode != current.StatusCode)
|
||||
{
|
||||
state.LastValues[tagRef] = current;
|
||||
_onChange(state.Handle, tagRef, current);
|
||||
if (_onBatch is not null) (changed ??= new List<DataChangeItem>()).Add(new DataChangeItem(tagRef, current));
|
||||
}
|
||||
}
|
||||
if (changed is { Count: > 0 }) _onBatch!(state.Handle, changed);
|
||||
```
|
||||
|
||||
Keep the surrounding contract-violation check and everything else byte-identical. Update the class remarks to document the batch callback (one raise per cycle, only when non-empty, after all per-tag callbacks of that cycle).
|
||||
|
||||
**Step 4: Run tests**
|
||||
|
||||
Run: `dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests` (whole project — the existing engine tests must stay green).
|
||||
Expected: PASS.
|
||||
|
||||
**Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add -A && git commit -m "feat(core): PollGroupEngine optional per-cycle batch callback"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3a: Modbus + S7 raise the batch event
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 3b, Task 5
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs`
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs`
|
||||
- Test: `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs` (create; mirror the fixture idiom of `ModbusSubscriptionTests.cs`)
|
||||
|
||||
**Step 1: Write the failing test** — subscribe via the driver against the existing in-test transport fake (copy `ModbusSubscriptionTests`' setup); assert:
|
||||
1. `ModbusDriver` is `IBatchSubscribable`.
|
||||
2. After one poll cycle with N changed registers, `OnDataChangeBatch` fired once with N items AND `OnDataChange` fired N times (both events, identical content — contract from Task 1).
|
||||
|
||||
**Step 2:** Run: `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests --filter "FullyQualifiedName~ModbusBatchSubscriptionTests"` — FAIL (interface not implemented).
|
||||
|
||||
**Step 3: Implement** — in each driver:
|
||||
- Change the class declaration to implement `IBatchSubscribable` (it already implements `ISubscribable`).
|
||||
- Add `public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;`
|
||||
- Find the `new PollGroupEngine(...)` construction; pass `onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items))`.
|
||||
|
||||
Per-tag raising is already wired via the engine's `onChange` — leave it.
|
||||
|
||||
**Step 4:** Run the Modbus + S7 test projects:
|
||||
`dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests && dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests` — PASS (all existing tests too).
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "feat(drivers): Modbus + S7 implement IBatchSubscribable via PollGroupEngine batch callback"`
|
||||
|
||||
---
|
||||
|
||||
### Task 3b: AbCip + FOCAS raise the batch event
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** Task 3a, Task 5
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs`
|
||||
- Modify: `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs`
|
||||
|
||||
Same mechanical change as Task 3a (interface + event + `onBatch` at the `new PollGroupEngine(...)` site). If either driver constructs the engine differently (check first), adapt but keep the identical contract. No new per-driver test required (the engine behavior is pinned by Task 2; the wiring pattern is pinned by Task 3a's test) — but the projects' existing suites must stay green.
|
||||
|
||||
Run: `dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests && dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests`
|
||||
|
||||
Commit: `feat(drivers): AbCip + FOCAS implement IBatchSubscribable`
|
||||
|
||||
---
|
||||
|
||||
### Task 4: DriverInstanceActor batch forwarding
|
||||
|
||||
**Classification:** high-risk (actor model, subscription lifecycle)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** Task 5
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorBatchForwardTests.cs` (create; copy the TestKit + stub-driver idiom from the existing `DriverInstanceActor*Tests` in that directory)
|
||||
|
||||
**Step 1: Write the failing tests**
|
||||
1. `Batch_driver_forwards_one_batch_message` — stub driver implementing `IBatchSubscribable`; raise `OnDataChangeBatch` with 3 items after subscribe; parent probe receives exactly ONE `AttributeValueBatchPublished` with 3 items (driver id + per-item FullReference/Value/StatusCode/timestamp mapped as `OnDataChangeForward` maps them today: `ts = Snapshot.SourceTimestampUtc ?? Snapshot.ServerTimestampUtc`).
|
||||
2. `Batch_driver_per_tag_event_not_double_forwarded` — the stub raises BOTH events (per contract); parent receives only the batch message, no legacy `AttributeValuePublished`.
|
||||
3. `Per_tag_driver_forwards_single_item_batches` — stub implementing only `ISubscribable`; one `OnDataChange` ⇒ parent receives one `AttributeValueBatchPublished` with 1 item.
|
||||
4. `Detach_removes_batch_handler` — after Unsubscribe (or Connected→Reconnecting, whichever the existing detach tests exercise), raising the batch event produces no message.
|
||||
|
||||
**Step 2:** Run: `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests --filter "FullyQualifiedName~DriverInstanceActorBatchForwardTests"` — FAIL.
|
||||
|
||||
**Step 3: Implement**
|
||||
- Add records next to `AttributeValuePublished` (`:109`):
|
||||
|
||||
```csharp
|
||||
/// <summary>One value within an <see cref="AttributeValueBatchPublished"/>.</summary>
|
||||
public sealed record AttributeValueItem(string FullReference, object? Value, uint StatusCode, DateTime TimestampUtc);
|
||||
/// <summary>Batched form of <see cref="AttributeValuePublished"/> — every changed attribute of ONE
|
||||
/// driver production burst in one message. The parent fans the whole batch to the mux and the OPC UA
|
||||
/// publish path in one hop each. Status codes ride verbatim (#519).</summary>
|
||||
public sealed record AttributeValueBatchPublished(string DriverInstanceId, IReadOnlyList<AttributeValueItem> Items);
|
||||
```
|
||||
|
||||
- Private message `private sealed record DataChangeBatchForward(IReadOnlyList<DataChangeItem> Items);` + field `private EventHandler<DataChangeBatchEventArgs>? _batchChangeHandler;`
|
||||
- In `HandleSubscribeAsync` (`:878-882`): when `subscribable is IBatchSubscribable batchable`, attach ONLY the batch handler (`_batchChangeHandler = (_, args) => self.Tell(new DataChangeBatchForward(args.Items)); batchable.OnDataChangeBatch += _batchChangeHandler;`) and do NOT attach `_dataChangeHandler`; else attach `_dataChangeHandler` exactly as today.
|
||||
- `Receive<DataChangeBatchForward>(OnDataChangeBatchForward);` beside the existing receive (`:480`); handler builds the items list once and Tells parent one `AttributeValueBatchPublished`.
|
||||
- `OnDataChangeForward` (`:1138`) becomes a single-item batch: `Context.Parent.Tell(new AttributeValueBatchPublished(_driverInstanceId, new[] { new AttributeValueItem(msg.FullReference, msg.Snapshot.Value, msg.Snapshot.StatusCode, ts) }));` — keep the `AttributeValuePublished` record declared (the mux comms-loss path and existing tests use the type), but the actor no longer sends it.
|
||||
- `DetachSubscription` (`:934-943`): also detach `_batchChangeHandler` from `IBatchSubscribable` and null it.
|
||||
|
||||
**IMPORTANT — downstream compile/behavior:** `DriverHostActor` currently receives `AttributeValuePublished` (`DriverHostActor.cs:960`). Until Task 8 lands, add a minimal bridge in THIS task so the system stays green: in `DriverHostActor`, add `Receive<DriverInstanceActor.AttributeValueBatchPublished>(batch => { foreach (var i in batch.Items) ForwardToMux(new DriverInstanceActor.AttributeValuePublished(batch.DriverInstanceId, i.FullReference, i.Value, i.StatusCode, i.TimestampUtc)); });` — Task 8 replaces this bridge with the real batched fan-out. Existing `DriverHostActor` tests that send `AttributeValuePublished` directly stay green (the receive stays).
|
||||
|
||||
**Step 4:** Run: `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests` — PASS (existing DriverInstanceActor forward tests may need updating IF they assert the parent receives `AttributeValuePublished` from a live driver event — update those to expect the single-item batch; do NOT weaken assertions on content).
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "feat(runtime): DriverInstanceActor forwards batched data changes"`
|
||||
|
||||
---
|
||||
|
||||
### Task 5: `ValueWrite` + sink `WriteValues` default method
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** Task 3a, 3b, 4
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs`
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredAddressSpaceSink.cs`
|
||||
- Test: `tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/AddressSpaceSinkBatchTests.cs` (create)
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `Default_WriteValues_loops_WriteValue` — a minimal fake implementing only the existing members; call `WriteValues` with 3 items; the fake's recorded `WriteValue` calls match in order/content.
|
||||
2. `Deferred_sink_forwards_WriteValues_to_inner_once` — a fake inner that overrides `WriteValues` (records one batch call); `DeferredAddressSpaceSink.WriteValues` reaches the inner's override (NOT 3 `WriteValue` calls).
|
||||
|
||||
**Step 3: Implement**
|
||||
|
||||
In `IOpcUaAddressSpaceSink.cs`:
|
||||
|
||||
```csharp
|
||||
/// <summary>One value write within a <see cref="IOpcUaAddressSpaceSink.WriteValues"/> batch.
|
||||
/// Same semantics per field as <see cref="IOpcUaAddressSpaceSink.WriteValue"/> (#519: status verbatim).</summary>
|
||||
public readonly record struct ValueWrite(
|
||||
string NodeId, object? Value, uint StatusCode, DateTime SourceTimestampUtc, AddressSpaceRealm Realm);
|
||||
```
|
||||
|
||||
Add to the interface (after `WriteValue`):
|
||||
|
||||
```csharp
|
||||
/// <summary>Write a batch of Variable values. Semantically identical to calling
|
||||
/// <see cref="WriteValue"/> per item; an SDK-backed sink overrides this to take its internal
|
||||
/// lock ONCE per batch instead of once per value (the hot-path win). The default implementation
|
||||
/// loops <see cref="WriteValue"/> so existing sinks and fakes behave unchanged.</summary>
|
||||
void WriteValues(IReadOnlyList<ValueWrite> values)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(values);
|
||||
for (var i = 0; i < values.Count; i++)
|
||||
{
|
||||
var v = values[i];
|
||||
WriteValue(v.NodeId, v.Value, v.StatusCode, v.SourceTimestampUtc, v.Realm);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
In `DeferredAddressSpaceSink`, add the forwarding override: `public void WriteValues(IReadOnlyList<ValueWrite> values) => _inner.WriteValues(values);`
|
||||
|
||||
**Step 4:** `dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests` — PASS.
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "feat(commons): batched WriteValues on the address-space sink"`
|
||||
|
||||
---
|
||||
|
||||
### Task 6: OtOpcUaNodeManager batched write under one lock
|
||||
|
||||
**Classification:** high-risk (SDK locking)
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** Task 7
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs` (WriteValue region, `:508-533`)
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkAddressSpaceSink.cs` (forward `WriteValues` to the manager)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerBatchWriteTests.cs` (create; copy the manager-construction idiom from an existing OpcUaServer.Tests file that instantiates `OtOpcUaNodeManager`)
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `WriteValues_lands_every_value` — materialise 3 variables (or rely on create-on-first-write), `WriteValues` a 3-item batch, read each variable's Value/StatusCode/Timestamp back via the manager's public surface.
|
||||
2. `WriteValues_creates_missing_variables` — same lazy-create semantics as `WriteValue`.
|
||||
|
||||
**Step 3: Implement**
|
||||
- Extract the body of `WriteValue`'s `lock (Lock)` block into `private void WriteValueLocked(string nodeId, ushort ns, string key, object? value, uint statusCode, DateTime ts)` (verbatim move).
|
||||
- `WriteValue` computes `ns`/`key` then `lock (Lock) { WriteValueLocked(...); }` — behavior identical.
|
||||
- Add:
|
||||
|
||||
```csharp
|
||||
/// <summary>Batched WriteValue: one Lock acquisition for the whole batch. Same per-item semantics.</summary>
|
||||
public void WriteValues(IReadOnlyList<ValueWrite> values)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(values);
|
||||
if (values.Count == 0) return;
|
||||
EnsureAddressSpaceCreated();
|
||||
lock (Lock)
|
||||
{
|
||||
for (var i = 0; i < values.Count; i++)
|
||||
{
|
||||
var v = values[i];
|
||||
WriteValueLocked(v.NodeId, NamespaceIndexForRealm(v.Realm), MapKey(v.Realm, v.NodeId), v.Value, v.StatusCode, v.SourceTimestampUtc);
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- In `SdkAddressSpaceSink`, override `WriteValues` forwarding to the manager's method (it currently forwards `WriteValue` 1:1 — mirror that).
|
||||
|
||||
**Step 4:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests` — PASS.
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "feat(opcuaserver): node manager batched WriteValues under one lock"`
|
||||
|
||||
---
|
||||
|
||||
### Task 7: OpcUaPublishActor batch message
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~4 min
|
||||
**Parallelizable with:** Task 6
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorBatchTests.cs` (create; copy the actor-construction idiom from `OpcUaPublishActorApplyFailureTests.cs`)
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `Batch_message_reaches_sink_as_one_WriteValues_call` — fake sink records `WriteValues` batches; send `AttributeValueBatchUpdate` with 3 items; sink saw one batch of 3.
|
||||
2. `Batch_increments_write_counter_per_item` — whatever surface exposes `_writes` (check for an existing status/metrics accessor in the actor's tests; if none, assert via the telemetry meter or skip the counter assertion and note it) — at minimum assert behavior parity: 3 items produce the same observable effect as 3 single `AttributeValueUpdate` messages.
|
||||
|
||||
**Step 3: Implement**
|
||||
- Add next to `AttributeValueUpdate` (`:52`):
|
||||
|
||||
```csharp
|
||||
/// <summary>Batched form of <see cref="AttributeValueUpdate"/>: every fan-out write of one driver
|
||||
/// production burst in one message — the sink takes its lock once for the whole batch.</summary>
|
||||
public sealed record AttributeValueBatchUpdate(IReadOnlyList<ValueWrite> Items);
|
||||
```
|
||||
|
||||
(`ValueWrite` comes from `ZB.MOM.WW.OtOpcUa.Commons.OpcUa` — already imported.)
|
||||
- Register `Receive<AttributeValueBatchUpdate>(HandleAttributeBatchUpdate);` beside `:288`.
|
||||
- Handler mirrors `HandleAttributeUpdate` (`:319-331`): try/catch WARN on throw, `_sink.WriteValues(msg.Items)`, then `_writes += msg.Items.Count` via `Interlocked.Add(ref _writes, msg.Items.Count)` and `OtOpcUaTelemetry.OpcUaSinkWrite.Add(msg.Items.Count, new KeyValuePair<string, object?>("kind", "value"))` — per-item counting (invariant 9).
|
||||
|
||||
**Step 4:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests --filter "FullyQualifiedName~OpcUaPublishActorBatch"` then the whole project — PASS.
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "feat(runtime): OpcUaPublishActor batched value updates"`
|
||||
|
||||
---
|
||||
|
||||
### Task 8: DriverHostActor batched fan-out (replaces the Task 4 bridge)
|
||||
|
||||
**Classification:** high-risk (the hottest routing seam)
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (needs 4 + 7; Task 9 lands with it)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs` (Task 9 content — same commit is fine if splitting is awkward, but prefer two commits)
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs` (create; copy the host-actor fixture idiom from `DriverHostActorNativeAlarmTests.cs`)
|
||||
|
||||
**Step 1: Failing tests**
|
||||
1. `Batch_fans_out_to_all_node_realm_refs_in_one_publish_message` — arrange the host with an applied composition where tag A maps to 1 raw + 1 UNS NodeId and tag B maps to 1 raw; send `AttributeValueBatchPublished` with A+B; the publish-actor probe receives exactly ONE `AttributeValueBatchUpdate` containing 3 `ValueWrite`s with correct NodeId/Realm/value/status/timestamp.
|
||||
2. `Unmapped_items_are_dropped_not_faulted` — batch with one unmapped ref: other items still land; no crash.
|
||||
3. `Batch_reaches_mux_once` — mux probe receives ONE batch message carrying all items (not N).
|
||||
|
||||
**Step 3: Implement**
|
||||
|
||||
Replace the Task 4 bridge with a real handler:
|
||||
|
||||
```csharp
|
||||
private void ForwardBatchToMux(DriverInstanceActor.AttributeValueBatchPublished msg)
|
||||
{
|
||||
_dependencyMux?.Tell(msg); // one envelope per burst; the mux filters per item (Task 9)
|
||||
|
||||
if (_opcUaPublishActor is null) return;
|
||||
|
||||
List<ValueWrite>? writes = null;
|
||||
foreach (var item in msg.Items)
|
||||
{
|
||||
if (_nodeIdByDriverRef.TryGetValue((msg.DriverInstanceId, item.FullReference), out var nodeIds))
|
||||
{
|
||||
writes ??= new List<ValueWrite>(msg.Items.Count * 2);
|
||||
foreach (var n in nodeIds)
|
||||
writes.Add(new ValueWrite(n.NodeId, item.Value, item.StatusCode, item.TimestampUtc, n.Realm));
|
||||
}
|
||||
else if (_log.IsDebugEnabled)
|
||||
{
|
||||
_log.Debug("DriverHost {Node}: no bound NodeId for ({Driver},{Ref}) — value dropped",
|
||||
_localNode, msg.DriverInstanceId, item.FullReference);
|
||||
}
|
||||
}
|
||||
if (writes is not null)
|
||||
_opcUaPublishActor.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.AttributeValueBatchUpdate(writes));
|
||||
}
|
||||
```
|
||||
|
||||
Keep `ForwardToMux` and its `Receive<AttributeValuePublished>` registration untouched (comms-loss fan + legacy tests). Register `Receive<DriverInstanceActor.AttributeValueBatchPublished>(ForwardBatchToMux);` in place of the Task 4 bridge.
|
||||
|
||||
**Step 4:** `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests` — PASS.
|
||||
|
||||
**Step 5:** `git add -A && git commit -m "feat(runtime): DriverHostActor batched value fan-out"`
|
||||
|
||||
---
|
||||
|
||||
### Task 9: DependencyMuxActor batch receive
|
||||
|
||||
**Classification:** small
|
||||
**Estimated implement time:** ~3 min
|
||||
**Parallelizable with:** Task 8 (same PR-let; do after 8's records exist)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`
|
||||
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/` — add a batch case beside the existing mux tests (find them via `CalculationDependencyFlowTests.cs` / grep `DependencyMuxActor` in that project)
|
||||
|
||||
**Step 1: Failing test** — register a subscriber for ref X; send a batch containing X and an uninteresting Y; subscriber receives exactly one `DependencyValueChanged` for X, nothing for Y.
|
||||
|
||||
**Step 3: Implement** — add:
|
||||
|
||||
```csharp
|
||||
Receive<DriverInstanceActor.AttributeValueBatchPublished>(OnAttributeValueBatchPublished);
|
||||
```
|
||||
|
||||
```csharp
|
||||
private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg)
|
||||
{
|
||||
foreach (var item in msg.Items)
|
||||
{
|
||||
if (!_byRef.TryGetValue(item.FullReference, out var subscribers) || subscribers.Count == 0) continue;
|
||||
var dep = new VirtualTagActor.DependencyValueChanged(item.FullReference, item.Value, item.TimestampUtc, item.StatusCode);
|
||||
foreach (var sub in subscribers) sub.Tell(dep);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Keep the per-item receive (comms-loss path).
|
||||
|
||||
**Step 4/5:** project tests green; `git commit -m "feat(runtime): dependency mux accepts batched publishes"`
|
||||
|
||||
---
|
||||
|
||||
### Task 10: Raise pinned-dispatcher throughput
|
||||
|
||||
**Classification:** trivial
|
||||
**Estimated implement time:** ~2 min
|
||||
**Parallelizable with:** Task 1
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Resources/akka.conf:112-116`
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs:301-312` (the inline HOCON copy)
|
||||
|
||||
Change `throughput = 1` → `throughput = 100` in BOTH places (they must match — the inline HOCON exists for consumers without akka.conf; see the comment at `ServiceCollectionExtensions.cs:301-303`). Update the comments: pinning alone guarantees the SDK's single-thread requirement; throughput only sets how many queued messages drain per scheduling run, so 100 removes a mailbox re-schedule round-trip per message with zero concurrency change.
|
||||
|
||||
Run: `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests --filter "FullyQualifiedName~ServiceCollectionExtensionsTests"`
|
||||
|
||||
Commit: `perf(runtime): raise opcua pinned dispatcher throughput 1 -> 100`
|
||||
|
||||
---
|
||||
|
||||
### Task 11: Re-key node-manager maps (kill per-write key allocations)
|
||||
|
||||
**Classification:** high-risk (38 call sites, mechanical but wide)
|
||||
**Estimated implement time:** ~5 min (compiler-driven)
|
||||
**Parallelizable with:** none (after Task 6 — same file)
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs`
|
||||
|
||||
**Step 1: Baseline** — run `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests` green BEFORE starting (this task relies on the existing suite as its net; it should need no new tests, but the whole suite must pass unchanged after).
|
||||
|
||||
**Step 2: Implement**
|
||||
- Add near `MapKey` (`:126-139`):
|
||||
|
||||
```csharp
|
||||
/// <summary>Zero-allocation realm-qualified map key. Replaces the formatted "ns=N;s=id" string key —
|
||||
/// the (realm, id) pair is the same identity without the per-write NodeId + string allocations.</summary>
|
||||
private readonly record struct RealmKey(AddressSpaceRealm Realm, string Id);
|
||||
```
|
||||
|
||||
- Change `MapKey(AddressSpaceRealm realm, string id)` to return `new RealmKey(realm, id)`; change `MapKey(NodeId nodeId)` to `new RealmKey(RealmOf(nodeId.NamespaceIndex), (string)nodeId.Identifier)` — `RealmOf` already exists (~`:123`); confirm its exact name/signature and that every `MapKey(NodeId)` call site only ever passes nodes from the two owned namespaces (they do — SDK-facing seams on owned handles). If `nodeId.Identifier` can be non-string on any call path (numeric-id nodes), guard with `as string ?? nodeId.Identifier?.ToString() ?? string.Empty`.
|
||||
- Change every map/set keyed by the old string key to `RealmKey`: `_variables`, `_folders`, `_alarmConditions`, `_historizedTagnames`, `_eventNotifierSources`, `_nativeAlarmNodeIds`, and the notifier-wiring bookkeeping — let the compiler find all of them (38 `MapKey` call sites).
|
||||
- Prefix scans (`RemoveEquipmentSubtree` ~`:2539-2596`, `InScope` usages): match on `key.Realm == realm && key.Id.StartsWith(prefix, ...)` — semantics identical (the old string key embedded the realm as `ns=N;`).
|
||||
- `ResolveNodeState(string key)` (~`:2240-2261`): change its parameter to `RealmKey`.
|
||||
- Do NOT change public API surface or any behavior — this is a pure key-representation change.
|
||||
|
||||
**Step 3:** `dotnet build ZB.MOM.WW.OtOpcUa.slnx` then `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests` and `dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests` — ALL green, zero test edits expected (tests drive the public surface). If a test pokes the key format directly, stop and reassess rather than bending the test.
|
||||
|
||||
**Step 4:** `git add -A && git commit -m "perf(opcuaserver): struct realm keys replace formatted NodeId string keys"`
|
||||
|
||||
---
|
||||
|
||||
### Task 12: Full verification + docs
|
||||
|
||||
**Classification:** standard
|
||||
**Estimated implement time:** ~5 min
|
||||
**Parallelizable with:** none (last)
|
||||
|
||||
**Files:**
|
||||
- Modify: `CLAUDE.md` (only if a statement it makes is now false — check the pinned-dispatcher/`throughput` mentions; keep edits surgical)
|
||||
- Modify: `docs/v2/Galaxy.Performance.md` — add a short "value-path batching" note (one paragraph: the batch seams, the invariants, the dispatcher change)
|
||||
|
||||
Steps:
|
||||
1. `dotnet build ZB.MOM.WW.OtOpcUa.slnx` — zero warnings introduced.
|
||||
2. `dotnet test ZB.MOM.WW.OtOpcUa.slnx` — run what the environment can (DB-backed suites need `10.100.0.35`; if unreachable, run every non-DB project under tests/Core, tests/Drivers (unit), tests/Server/Runtime.Tests, tests/Server/OpcUaServer.Tests and record which suites were skipped and why in the commit message).
|
||||
3. Docs + commit: `docs: record value-pipeline batching (plan 2026-08-15)`.
|
||||
|
||||
---
|
||||
|
||||
## Explicit non-goals (do NOT let scope creep into these)
|
||||
|
||||
- **No conflation / bounded mailboxes** — batching per burst is this plan's backpressure story; keyed conflation is a follow-up.
|
||||
- **No fan-out-map push-down to DriverInstanceActor** — the host stays the router.
|
||||
- **No MQTT / MTConnect / Sparkplug / Galaxy / OpcUaClient batch implementations** — they flow through the single-item-batch path unchanged.
|
||||
- **No change to alarm paths** (`AttributeAlarmPublished`, `AlarmStateUpdate`) — value path only.
|
||||
- **No PollGroupEngine cadence change** (fixed-rate scheduling is a separate item).
|
||||
@@ -0,0 +1,19 @@
|
||||
{
|
||||
"planPath": "docs/plans/2026-08-15-value-pipeline-batching.md",
|
||||
"tasks": [
|
||||
{"id": 1, "subject": "Task 1: Batch contract types in Core.Abstractions", "status": "pending"},
|
||||
{"id": 2, "subject": "Task 2: PollGroupEngine batch callback", "status": "pending", "blockedBy": [1]},
|
||||
{"id": 3, "subject": "Task 3a: Modbus + S7 raise the batch event", "status": "pending", "blockedBy": [2]},
|
||||
{"id": 4, "subject": "Task 3b: AbCip + FOCAS raise the batch event", "status": "pending", "blockedBy": [2]},
|
||||
{"id": 5, "subject": "Task 4: DriverInstanceActor batch forwarding", "status": "pending", "blockedBy": [1]},
|
||||
{"id": 6, "subject": "Task 5: ValueWrite + sink WriteValues default method", "status": "pending"},
|
||||
{"id": 7, "subject": "Task 6: OtOpcUaNodeManager batched write under one lock", "status": "pending", "blockedBy": [6]},
|
||||
{"id": 8, "subject": "Task 7: OpcUaPublishActor batch message", "status": "pending", "blockedBy": [6]},
|
||||
{"id": 9, "subject": "Task 8: DriverHostActor batched fan-out", "status": "pending", "blockedBy": [5, 8]},
|
||||
{"id": 10, "subject": "Task 9: DependencyMuxActor batch receive", "status": "pending", "blockedBy": [9]},
|
||||
{"id": 11, "subject": "Task 10: Raise pinned-dispatcher throughput", "status": "pending"},
|
||||
{"id": 12, "subject": "Task 11: Re-key node-manager maps", "status": "pending", "blockedBy": [7]},
|
||||
{"id": 13, "subject": "Task 12: Full verification + docs", "status": "pending", "blockedBy": [3, 4, 9, 10, 11, 12]}
|
||||
],
|
||||
"lastUpdated": "2026-08-15T00:00:00Z"
|
||||
}
|
||||
Reference in New Issue
Block a user