From 1040dc0fcc4a4a0fea9673c7db218e1b16a26b76 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 14 Aug 2026 20:30:35 -0400 Subject: [PATCH] =?UTF-8?q?docs(plans):=20DCL=20batch=20seam=20design=20?= =?UTF-8?q?=E2=80=94=20WP2.1a?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plans/2026-08-14-dcl-batch-seam-design.md | 103 ++++++++++++++++++ 1 file changed, 103 insertions(+) create mode 100644 docs/plans/2026-08-14-dcl-batch-seam-design.md diff --git a/docs/plans/2026-08-14-dcl-batch-seam-design.md b/docs/plans/2026-08-14-dcl-batch-seam-design.md new file mode 100644 index 00000000..ca1ff105 --- /dev/null +++ b/docs/plans/2026-08-14-dcl-batch-seam-design.md @@ -0,0 +1,103 @@ +# DCL Batch Seam Design — WP2.1a + +**Date:** 2026-08-14 · **Scope:** design contract for WP2.1b (implementation) · **Findings:** arch-review #3 (High) + MxGateway 2-RPC subscribe / no-backoff retry (Med) + quality-counter and alarm-fanout costs (Med) · **Sized for:** 37,500 tags/site (500 instances × 75 tags, `docs/deployment/topology-guide.md:255`). + +## 1. Batch API on `IDataConnection` + +```csharp +public record TagSubscribeResult(string TagPath, bool Success, string? SubscriptionId, string? ErrorMessage); + +Task> SubscribeBatchAsync( + IReadOnlyList tagPaths, SubscriptionCallback callback, CancellationToken ct = default); +Task UnsubscribeBatchAsync(IReadOnlyList subscriptionIds, CancellationToken ct = default); +``` + +- **One shared callback, plain tag list** — deliberately simpler than the plan's sketched `IReadOnlyList` input record. The only caller passes an identical closure for every tag (`DataConnectionActor.HandleSubscribe`, `DataConnectionActor.cs:745-748`: `(path, value) => self.Tell(new TagValueReceived(path, value, generation))`), and `SubscriptionCallback` already delivers the tag path. A per-tag record would carry nothing. +- **Partial-failure contract:** per-tag faults (bad node id, resolution failure) come back as result rows with `Success:false` and never throw — mirroring the gateway's `SubscribeResult`/`BulkSubscribeReply` (mxaccess_gateway.proto:632) and OPC UA per-item create status. A **thrown** exception means the whole batch failed at connection level; the actor classifies it with the existing `IsConnectionLevelFailure` (`DataConnectionActor.cs:1056`) and drives Reconnecting, exactly as the per-tag path does today (`SubscribeTagResult.ConnectionLevelFailure`, line 758-761). +- **`ReadBatchAsync`/`WriteBatchAsync` keep their existing signatures** (`IDataConnection.cs:47,58` — dictionary-of-results, already right) but the contract is strengthened from "may loop over singles" to **true bulk**: `OpcUaDataConnection` currently loops per tag (`OpcUaDataConnection.cs:250-274, 289-315`); it becomes one `ReadValueIdCollection`/`WriteValueCollection` service call (the MxGateway adapter is the in-repo reference — `MxGatewayDataConnection.cs:267-290` already delegates to `ReadBulkAsync`/`WriteBulkAsync`). Adapters chunk internally at 1,000 items per service call to stay under typical OPC UA `MaxNodesPerRead/Write` operation limits. Preserved semantics: per-tag failure rows, `OperationCanceledException` aborts the whole batch (`OpcUaDataConnection.cs:263-267`). +- **Single-tag methods are KEPT**, implemented as batch-of-one delegations. They have real callers outside the hot path: the OPC UA heartbeat monitor (`OpcUaDataConnection.cs:145`), `WriteBatchAndWaitAsync`'s poll loop (line 355), `HandleWrite`/`HandleReadTagValues`. Removing them churns every test fake for zero gain; the actor's hot paths simply stop using them. + +## 2. OPC UA: sharding + apply serialization + +- **Budget option:** `OpcUaEndpointConfig.MaxMonitoredItemsPerSubscription`, default **5000**, flowing into `OpcUaConnectionOptions` like the other subscription knobs (`OpcUaDataConnection.cs:74-96`). Per-endpoint, not global — item-count ceilings are a property of the target server. 37,500 tags → 8 shards. +- `RealOpcUaClient` replaces the single `_subscription` (`RealOpcUaClient.cs:22,150-162`) with a shard list. Placement: first shard with free capacity, else create a new shard from the same options; a shard is deleted when it empties. A `handle → shard` map serves removal and keeps `RemoveSubscriptionsAsync` grouping one `ApplyChangesAsync` per touched shard. +- **Alarm (A&C) monitored items get a dedicated event shard.** `TriggerConditionRefreshAsync` passes a subscription id (`RealOpcUaClient.cs:784-786`); pinning event items to one shard keeps that bookkeeping trivial and keeps `QueueSize:1000` event items from occupying data-shard capacity. +- **Batch subscribe = N×`AddItem` + one `ApplyChangesAsync` per touched shard per chunk** (today it is one Apply per tag — `RealOpcUaClient.cs:494-495`, the core of finding #3). New `IOpcUaClient` members: `CreateSubscriptionsAsync`, `RemoveSubscriptionsAsync`, `ReadValuesAsync`, `WriteValuesAsync`, all with per-item results. +- **All `ApplyChangesAsync`/`CreateAsync`/`DeleteAsync` calls serialize behind ONE `SemaphoreSlim(1,1)` per client.** The SDK Subscription is not safe under concurrent structural mutation, and today nothing prevents `HandleSubscribe`'s background task, the resolution-retry tick, and an alarm subscribe from applying concurrently. Batching makes contention rare; the single lock (rather than per-shard) also serializes shard creation and is the simplest thing that is correct. + +## 3. Reconnect orchestration & re-seed + +Two distinct paths, deliberately staggered differently: + +- **Instance-driven subscribes** (site failover): DeploymentManager already staggers instance creation (`SiteRuntimeOptions.StartupBatchSize` 20 / `StartupBatchDelayMs` 100, `SiteRuntimeOptions.cs:13-19`), so `SubscribeTagsRequest`s arrive pre-spaced at ~75 tags each. Each request becomes ONE `SubscribeBatchAsync` (a single chunk). **The DCL adds no extra delay on this path** — double-staggering would only slow failover. +- **Adapter-level reconnect** (`ReSubscribeAll`, `DataConnectionActor.cs:1600-1684`): today it fires 37,500 independent `SubscribeAsync` tasks in a tight loop (lines 1642-1653). It becomes sequential chunks of **`SubscribeBatchSize` = 500** with **`SubscribeBatchDelay` = 50 ms** between chunks, all inside one background task carrying the captured adapter + generation; each chunk pipes one completion message (per-tag results inside) back to the actor. 75 chunks ≈ 15-25 s at realistic RTTs — versus today's unbounded task storm. Sequential (not parallel) chunks: subscription creation on one session is serialized by the apply lock anyway, and sequencing bounds device load. +- **`SeedTagsAsync`** (`DataConnectionActor.cs:800-851`) currently reads **per tag, serially**, each with a 30 s timeout — the 30 s-per-tag worst case the plan calls out. It becomes chunked `ReadBatchAsync` calls: **`SeedReadBatchSize` = 250**, **`SeedReadMaxParallelism` = 4** concurrent chunks, `SeedReadTimeout` (existing, 30 s) now applying **per chunk**, under an **overall deadline `SeedOverallTimeout` = 120 s** enforced by one linked CTS across all chunks and retry rounds. The existing retry semantics survive unchanged in shape: still-empty tags re-read up to `SeedReadMaxAttempts` (3) with `SeedReadRetryDelay` (250 ms) between rounds; on deadline expiry the remaining tags are logged and stay Uncertain until a change notification — the exact behavior documented at `DataConnectionOptions.cs:17-25`. Chunk size stays moderate on purpose: the "some gateways time out on a large batch" caveat that motivated per-tag reads (`DataConnectionActor.cs:792-793`) is answered by chunking + per-chunk timeout, not by giving up bulk. Sizing: 150 chunks / 4-way ≈ 38 waves; ~20-40 s typical, so 120 s has honest headroom. +- All new knobs live on `DataConnectionOptions` (validated in `DataConnectionOptionsValidator`): `SubscribeBatchSize`, `SubscribeBatchDelay`, `SeedReadBatchSize`, `SeedReadMaxParallelism`, `SeedOverallTimeout`. + +## 4. Tag-resolution retry backoff + +Today: fixed 10 s periodic timer, one `SubscribeAsync` task per unresolved tag per tick (`DataConnectionActor.cs:1536-1574`). A dead device with thousands of unresolved tags probes forever at full width. + +- **Exponential backoff:** interval starts at `TagResolutionRetryInterval` (existing, 10 s), doubles per fully-failed round, capped at **`TagResolutionRetryMaxInterval` = 5 min** (new option). Reset to the floor when any tag resolves or on reconnect (`ReSubscribeAll` already clears `_unresolvedTags`, line 1617). +- **Batched probes:** each tick issues `SubscribeBatchAsync` over the not-in-flight unresolved set, chunked at `SubscribeBatchSize` — not N single subscribes. `_resolutionInFlight` dedup survives at chunk granularity. +- **Single-shot timer, rescheduled on probe completion** — a periodic timer cannot back off. This inherently preserves the anti-starvation property the current code defends with its `IsTimerActive` gate (`DataConnectionActor.cs:1015-1028`): the next tick is scheduled only when the previous round's completion message arrives, so a fan-out of failures can never reset the clock. + +## 5. MxGateway strategy — no cross-repo change needed for the main path + +Checked `~/Desktop/MxAccessGateway`: the proto **already has** the bulk family — `SubscribeBulkCommand` (AddItem + Advise per tag in one worker STA pass, one gRPC round trip; `mxaccess_gateway.proto:325`, worker `MxAccessSession.SubscribeBulk`), `UnsubscribeBulkCommand`, `AddItemBulkCommand`, `ReadBulkCommand`, `WriteBulkCommand`, all with per-item results. The **shipped `ZB.MOM.WW.MxGateway.Client` 0.2.0** — the exact version this repo pins (`Directory.Packages.props:98`) — exposes `SubscribeBulkAsync`/`AddItemBulkAsync`/`UnsubscribeBulkAsync` (verified in the packaged DLL, `dist/ZB.MOM.WW.MxGateway.Client.0.2.0.nupkg`). So: + +- **Plain-advise mode** (a configured write-user): `SubscribeBatchAsync` = chunked `SubscribeBulkAsync` — the 2-RPC-per-tag subscribe (`RealMxGatewayClient.cs:93-103`: `AddItemAsync` + `AdviseAsync`) collapses to one RPC per 500 tags. Unsubscribe = `UnsubscribeBulkAsync`. +- **Supervisory mode** (`WriteUserId == 0`, `RealMxGatewayClient.cs:52`): the worker has **no bulk supervisory advise** (`SubscribeBulk` issues plain `Advise`; only single `AdviseSupervisory` exists). Fallback per plan: one `AddItemBulkAsync` for the chunk, then client-side **pipelined** per-item `AdviseSupervisory` commands with bounded parallelism **`MxSupervisoryAdviseParallelism` = 16** (new option) — 37,500 tags ≈ 2,344 in-flight windows instead of 75,000 serial round trips. The existing `Lazy` advise-once map (`RealMxGatewayClient.cs:47`) is pre-populated from the bulk path so a concurrent write still awaits the same advise. +- **Cross-repo follow-up (note, not a blocker):** add an additive `bool supervisory` field to `SubscribeBulkCommand` (or an `AdviseSupervisoryBulkCommand`) in mxaccessgw so supervisory sites also get one-RPC subscribe. File against mxaccessgw; ScadaBridge picks it up on the next client bump with a capability/`Unimplemented` fallback to the pipelined path. +- Read/Write are already true bulk in this adapter — no change. + +## 6. Quality-counter flush + +`HandleTagValueReceived` calls `_healthCollector.UpdateTagQuality` on **every** value received, even when quality is unchanged (`DataConnectionActor.cs:1811`). New policy: in-actor bucket math stays per-message (cheap), but the collector push is **transition-gated + coalesced** — skip entirely when `prevQuality == newQuality`; on a genuine transition set a dirty flag and flush on a **`QualityFlushInterval` = 1 s** single-shot timer. Health reports poll at 30 s, so 1 s coalescing loses nothing. **Immediate synchronous flushes are retained** where correctness depends on them: `PushBadQualityForAllTags` on disconnect (line 1595), unsubscribe (line 1135), and the `ReSubscribeAll` reset (line 1635). + +## 7. Alarm subscriber prefix index + gateway union filter + +- **Index:** `HandleAlarmTransitionReceived` scans every subscribed source and does two `StartsWith` per source per transition (`DataConnectionActor.cs:1974-1979`). Replace with buckets keyed by the source reference's **first path segment** (split on `.`): lookup extracts the first segment of `SourceObjectReference`/`SourceReference` and tests only that bucket. Sources whose reference contains no separator (a prefix shorter than one full segment) go to a small linear residue list — exact semantics preserved, including the per-source condition-type gate and the `SnapshotComplete` broadcast-to-all (lines 1965-1972), which bypasses the index entirely. +- **Gateway union filter:** the MxGateway alarm stream is opened **gateway-wide** today (`alarmFilterPrefix: null`, `MxGatewayDataConnection.cs:231-236`). `StreamAlarmsRequest` carries a single `alarm_filter_prefix` (`mxaccess_gateway.proto:986`), so the pushable union is the **longest common prefix** of the active source references. The adapter recomputes the LCP on subscribe; the stream restarts (cancel + reopen — the source replays a fresh snapshot, which `NativeAlarmActor` already handles) **only when a new source falls outside the current prefix**. Unsubscribes never restart — a too-broad prefix is only bandwidth. The prefix stays a bandwidth optimisation, never correctness: the actor's per-source + condition-type gate remains authoritative, mirroring the OPC UA WhereClause stance (`RealOpcUaClient.cs:659-662`). + +## 8. Behaviors that MUST be preserved (stage-b checklist) + +1. **Become/Stash lifecycle** — Connecting/Connected/Reconnecting states, stash rules per state (`DataConnectionActor.cs:249-550`) unchanged. +2. **Generation fencing** — batch callbacks capture `_adapterGeneration` exactly as the per-tag closures do (`:697,1637`); `TagValueReceived`/`AlarmTransitionReceived` still drop stale generations (`:1771,1949`). Chunked reconnect runs against the adapter captured at chunk-dispatch time (S7 discipline, `:698-704`). +3. **Immediate bad quality on disconnect** — `PushBadQualityForAllTags` stays synchronous in `BecomeReconnecting` (`:473`), never deferred to the quality-flush timer. +4. **Transparent re-subscribe** — derived from `_subscriptionsByInstance` (durable truth, `:1602-1607`), `_instancesByTag` NOT cleared (`:1612-1615`), reconnect re-seed included (`:1655-1683`). +5. **Write failures synchronous to the calling script** — `HandleWrite`/`HandleWriteBatch` `PipeTo` with `WriteTimeout` translation unchanged (`:1141-1242`). +6. **Seed-after-registration** — seeds still ride `SubscribeCompleted` and are delivered only after `_subscriptionsByInstance` registration (the static-tag fix, `:765-782,999-1012`); only resolved tags are seeded. +7. **In-flight orphan guards** — unsubscribe-during-subscribe releases adapter handles (`:860-904`), duplicate alarm feeds released not overwritten (`:1908-1932`); batch completions apply per-tag through the same guard logic. +8. **Unresolved-tag semantics** — Bad-quality `TagValueUpdate` signal (`:991-995`), `_subscribesInFlight`/`AlreadySubscribed` dedup (`:706-729`), counter promotion rules (`:938-996`). +9. **Alarm filter last-writer-wins warning** (`:1841-1848`) and the client-side gate as sole authority. + +## 9. Stage-b test plan (fake-client) + +Existing DCL suite (`tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/` — actor lifecycle, generation fencing, alarm routing) must stay green. New pins: + +- **One apply per chunk:** fake `IOpcUaClient` counts `ApplyChangesAsync` calls — N tags through `SubscribeBatchAsync` → `ceil(N/SubscribeBatchSize)` applies, not N. +- **Sharding:** 12,001 tags with budget 5,000 → 3 shards; removal empties → shard deleted; alarm item lands on the event shard and `ConditionRefresh` targets it. +- **Apply serialization:** fake apply with an injected delay + concurrency flag — concurrent batch subscribe / unsubscribe / alarm subscribe never overlap an apply. +- **Re-seed deadline:** fake `ReadBatchAsync` that hangs → seed returns at `SeedOverallTimeout`, remaining tags pending/Uncertain, `SubscribeCompleted` still delivered and acked. +- **Backoff caps:** all-fail probe rounds produce intervals 10 s, 20 s, 40 s … capped at `TagResolutionRetryMaxInterval` (compressed values in-test); any success resets to the floor; probes are batched (one client call per chunk, not per tag). +- **MxGateway modes:** plain mode → exactly one `SubscribeBulkAsync` per chunk; supervisory mode → one `AddItemBulkAsync` + per-item advises with in-flight count ≤ `MxSupervisoryAdviseParallelism`. +- **Partial failure surface:** a per-tag failure row → `_unresolvedTags` + Bad-quality push, response `Success:true`; a thrown batch → all-tags connection-level → `SubscribeTagsResponse(Success:false)` + Reconnecting. +- **Quality flush:** M updates at unchanged quality → zero collector calls; one transition → one flush within the interval; disconnect flush still immediate. +- **Alarm index equivalence:** routing decisions identical to the linear scan over a sample set including shared-prefix sources, sub-segment prefixes (residue list), and the empty-source `SnapshotComplete` broadcast. LCP stream restarts only when a new source escapes the current prefix. + +## 10. Defaults summary + +| Option | Home | Default | +|---|---|---| +| `MaxMonitoredItemsPerSubscription` | `OpcUaEndpointConfig` | 5000 | +| `SubscribeBatchSize` | `DataConnectionOptions` | 500 | +| `SubscribeBatchDelay` | `DataConnectionOptions` | 50 ms | +| `SeedReadBatchSize` | `DataConnectionOptions` | 250 | +| `SeedReadMaxParallelism` | `DataConnectionOptions` | 4 | +| `SeedOverallTimeout` | `DataConnectionOptions` | 120 s | +| `TagResolutionRetryMaxInterval` | `DataConnectionOptions` | 5 min | +| `QualityFlushInterval` | `DataConnectionOptions` | 1 s | +| `MxSupervisoryAdviseParallelism` | `DataConnectionOptions` | 16 | + +Existing options keep their meanings; `SeedReadTimeout` (30 s) becomes per-chunk rather than per-tag.