Joseph Doherty
104 lines
17 KiB
Markdown
104 lines
17 KiB
Markdown
# 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<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
|
||
IReadOnlyList<string> tagPaths, SubscriptionCallback callback, CancellationToken ct = default);
|
||
Task UnsubscribeBatchAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default);
|
||
```
|
||
|
||
- **One shared callback, plain tag list** — deliberately simpler than the plan's sketched `IReadOnlyList<TagSubscription>` 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<Task>` 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.
|