Compare commits

..

15 Commits

Author SHA1 Message Date
Joseph Doherty a346d514dd test(contracts): scope command-reply fixture invariants past the CLI-40/41 authenticate-user malformed-reply fixtures
ci / portable (push) Successful in 14m0s
ci / java (push) Successful in 6m50s
ci / windows-x86 (push) Failing after 1m21s
ci / nightly-windev (push) Has been skipped
The blanket loop asserted HRESULT/Statuses/ReturnValue on every command_replies fixture, but the authenticate-user.* fixtures added for the malformed-reply and credential-redaction contracts deliberately omit them (NRE on ReturnValue.DataType). Keep universal Kind/ProtocolStatus invariants for all; apply the MXAccess-detail block only to fixtures that carry it. Test-only.
2026-08-07 08:48:49 -04:00
Joseph Doherty a2d3f66b8b docs(archreview): record next-cycle candidate findings + pending operator actions surfaced during remediation
ci / windows-x86 (push) Successful in 1m19s
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m19s
ci / portable (push) Failing after 4m55s
2026-08-07 08:48:03 -04:00
Joseph Doherty 93d84019b9 docs(tracking): sync IPC-23 domain register to Done (doc wave landed; Grpc.md row intentionally scoped out — DrainEvents is a worker diagnostic)
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m19s
ci / java (push) Successful in 2m4s
ci / portable (push) Failing after 4m43s
2026-08-07 08:47:31 -04:00
Joseph Doherty 6d26ed094c docs(tracking): close old-tracker CLI-24, CLI-34 as Done (2026-07-12 review old-tracker actions; both incidentally fixed)
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m22s
ci / windows-x86 (push) Successful in 1m20s
ci / portable (push) Failing after 4m26s
2026-08-07 08:11:57 -04:00
Joseph Doherty 4201da63d2 docs(tracking): flip IPC-24/IPC-25 to Done in the Contracts&IPC domain register (missed by the codegen-wave tracker update)
ci / windows-x86 (push) Failing after 1m19s
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m17s
ci / portable (push) Failing after 4m51s
2026-08-07 08:10:39 -04:00
Joseph Doherty 9c780f8164 Merge branch 'fix/cli-39-version-train'
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m20s
ci / java (push) Successful in 2m12s
ci / portable (push) Failing after 4m44s
# Conflicts:
#	archreview/2026-07-12/remediation/00-tracking.md
2026-08-07 08:09:39 -04:00
Joseph Doherty 440e7cf03d fix(CLI-39): bump Contracts nupkg to 0.2.0; scope pack-clients.ps1 regexes
Code review of the CLI-39 branch caught an Important gap: Contracts.csproj
was left at the already-published 0.1.2 while the .NET Client moved to
0.2.0. Invoke-PackDotnet in scripts/pack-clients.ps1 packs and publishes
both ZB.MOM.WW.MxGateway.Contracts and .Client through the same -Publish
loop, and the new collision guard runs every nupkg it finds through
Assert-GiteaPackageNotPublished. Left as-is, the next real .NET publish
would pack Contracts at 0.1.2, the guard would correctly refuse to
republish it, and the loop would abort mid-way with Client (alphabetically
first) possibly already pushed -- the two packages permanently out of
lockstep.

- src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj:
  <Version> 0.1.2 -> 0.2.0, matching the .NET Client (they have always
  released together).
- src/Directory.Build.props: corrected a comment that was now stale --
  it claimed the repo-wide 0.1.2 default was kept to match the Contracts
  package, which is no longer true now that Contracts.csproj overrides it.
  The <Version> value itself is unchanged; Server/Worker/Tests staying at
  0.1.2 is a separate, not-yet-made decision, out of scope for CLI-39.
- docs/ClientPackaging.md: Contracts.csproj added as a fifth manifest in
  the Versioning section, with the near-miss recorded.

Also hardened scripts/pack-clients.ps1 per the same review: the Python
(pyproject.toml) and Rust (Cargo.toml) version-extraction regexes now
scope to the [project]/[package] section header instead of matching the
first "version = ..." line anywhere in the file (Cargo.toml has an
identical second one under [workspace.package] -- matching whichever came
first was luck of ordering, not correctness). One-line comment added on
the nuget filename-parse regex.

Verified live against the real Gitea registry: Contracts and Client both
still refuse at 0.1.2 and both now pass at 0.2.0, including running the
actual Invoke-PackDotnet filename-parse-then-guard logic against two
freshly packed real .nupkg files. dotnet build of Contracts.csproj and the
client slnx both clean. No publish performed.
2026-08-07 08:07:40 -04:00
Joseph Doherty ae605d2368 Merge remote-tracking branch 'origin/fix/wrk-22-25-seam'
ci / windows-x86 (push) Failing after 1m19s
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m10s
ci / portable (push) Failing after 4m27s
# Conflicts:
#	archreview/2026-07-12/remediation/00-tracking.md
#	archreview/2026-07-12/remediation/30-contracts-ipc.md
2026-08-07 08:01:10 -04:00
Joseph Doherty 9b2abef4e1 fix(CLI-39): bump client versions off published 0.1.2; guard the publish pipeline
Converges all five clients on one version after four had drifted onto the
already-published 0.1.2/0.1.1 while their APIs kept changing underneath it:

- Rust Cargo.toml [package] + [workspace.package] -> 0.2.0 (CLIENT_VERSION
  already derives from CARGO_PKG_VERSION, no separate edit).
- Python pyproject.toml + version.py -> 0.2.0; new test asserts __version__
  matches pyproject.toml (closes the CLI-26 residual drift mode).
- Go mxgateway/version.go ClientVersion -> 0.2.0.
- .NET ZB.MOM.WW.MxGateway.Client.csproj <Version> -> 0.2.0.
- Java -> 0.2.1, not 0.2.0: the live Gitea Maven feed already had 0.2.0
  published (2026-06-26), before the CLI-37/38/40/41 conformance fixes
  changed the client's observable behavior, so reusing 0.2.0 would label
  two different APIs identically. Recorded as an exception in
  docs/ClientPackaging.md's new Versioning section.

Publish-pipeline guards:

- scripts/tag-go-module.ps1 implements the CLI-21 guard: after semver
  validation it refuses to tag unless clients/go/mxgateway/version.go's
  ClientVersion already matches the requested tag version.
- scripts/pack-clients.ps1 gains a Gitea package-registry collision guard
  wired into every per-language -Publish step; it aborts if the target
  name+version already exists rather than force-overwriting. Verified live
  against the real Gitea registry (credentials already present in this
  environment) — correctly refuses on every known-published artifact and
  passes on every unpublished target.

Docs updated in the same commit: docs/ClientPackaging.md (new Versioning
section), and the five client READMEs' stale 0.1.1/0.1.2 example versions.

No .proto changes. No publish performed.
2026-08-07 07:58:49 -04:00
Joseph Doherty 815e58d28b docs(tracking): record WRK-22/24/25/27 + IPC-26 windev evidence (377 pass)
ci / java (push) Successful in 2m7s
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m12s
ci / portable (push) Failing after 4m12s
2026-08-07 07:52:44 -04:00
Joseph Doherty 8df35cd63a fix(WRK-22,WRK-24,WRK-25,WRK-27,IPC-26): worker write-seam hardening
ci / java (push) Successful in 2m7s
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m13s
ci / portable (push) Failing after 4m6s
WRK-22/IPC-26: tombstone a WriteAsync/WriteBatchAsync cancelled while
waiting for the write lock (PendingFrame.Claimed under _gate; DequeueNext
skips cancelled, claims the frame it returns) so a cancelled write never
reaches the wire unless already claimed mid-write (documented residual).

WRK-25: add WriteBatchAsync; RunEventDrainLoopAsync submits the drained
event batch through it, so a burst of N events costs one flush not N.
IPC-30 oversized-event structured fault preserved via FindOversizedEvent.

WRK-24: reject a below-1024 negotiated frame maximum at the handshake
(MinNegotiableFrameBytes, matching GatewayOptionsValidator floor).

WRK-27: alarm poll advertises StaCallInProgress on the heartbeat snapshot
so the watchdog suppresses to the ceiling, not the grace.

Docs (WorkerFrameProtocol.md, MxAccessWorkerInstanceDesign.md) and the
2026-07-12 remediation registers/change-log updated in the same commit.
2026-08-07 07:50:38 -04:00
Joseph Doherty a55956ffa5 Merge branch 'fix/tst-30-runner-docs'
ci / java (push) Successful in 2m11s
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m38s
ci / portable (push) Failing after 17m43s
# Conflicts:
#	archreview/2026-07-12/remediation/00-tracking.md
2026-08-07 07:50:21 -04:00
Joseph Doherty aba22358f5 Merge branch 'fix/ipc-27-descriptor-test'
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Successful in 1m14s
ci / java (push) Successful in 2m8s
ci / portable (push) Failing after 4m45s
2026-08-07 07:49:33 -04:00
Joseph Doherty b604fed72b docs(TST-30): document shared-runner CI bottleneck + second-runner operator runbook
Doc half of TST-30 (single shared Gitea runner is a CI throughput/availability
bottleneck): docs/GatewayTesting.md's Continuous Integration section gains a
"Runner capacity is shared and finite" subsection covering the maxParallel=1
instance-level runner shared with dohertj2/lmxopcua, the ~20-30 min queue
latency observed under cross-repo contention, and Gitea 1.26's missing run
cancel/delete API. The existing "windev tier down" degraded-mode paragraph now
also covers "runner contended" as a reason to bypass the queue via
CI_SHA=<sha> scripts/ci/run-windev-ci.sh <mode> or the manual windev worktree
flow, generalizing it per the finding's design note.

New operator runbook docs/runbooks/TST-30-second-ci-runner.md carries the
actual runner registration (option a: second act_runner instance on
10.100.0.35 with the same container.network: traefik config, recommended;
option b: dedicated labelled runner, escalation only; option c: runner on
windev, rejected) plus verification steps and the no-cancel caveat. The
optional workflow-level concurrency group is documented as unverified --
framed as "verify before relying on it" -- and left unimplemented in ci.yml,
since registering the runner and any runs-on gating is operator/infra work
outside this repo's tree.

Tracking: TST-30 -> Done (doc half; runner registration operator-pending) in
both registers + change-log row.
2026-08-07 07:47:37 -04:00
Joseph Doherty 6060d21995 fix(IPC-27): close descriptor-freshness blind spots for enums, services, and Galaxy
ClientProtoInputTests.Descriptor_ContainsEveryContractMessageAndField only
compared messages and fields, and only enumerated the gateway/worker
descriptors, so a new enum value, a new RPC, or any galaxy_repository.proto-only
change would not redden the test even though it is documented as the primary
protoc-free CI gate.

Rename to Descriptor_ContainsEveryContractSymbol and extend the reflection walk
on both sides (published protoset and in-process contract) to also collect
enums/enum values ({enumFullName}, {enumFullName}/{valueName}) and
services/methods ({serviceFullName}, {serviceFullName}/{methodName}), and add
GalaxyRepositoryReflection.Descriptor to the enumerated files. The comparison
stays a flat, order-insensitive string-set diff with no protoc dependency.

Update docs/ClientProtoGeneration.md and docs/Contracts.md prose from
"message or field" to the full symbol coverage.

Red-path proof: pointed the test at the pre-IPC-01 stale protoset and confirmed
it failed naming max_frame_bytes, several MxCommandKind/AlarmProviderMode enum
values, MxAccessGateway/StreamAlarms and GalaxyRepository/BrowseChildren, and
the galaxy_repository.v1.* surface; restored the real path and re-ran green.

Flips IPC-27 to Done in the 2026-07-12 remediation tracker and register.
2026-08-07 07:47:18 -04:00
43 changed files with 1481 additions and 120 deletions
File diff suppressed because one or more lines are too long
@@ -17,12 +17,12 @@ members, no positional records). The worker builds and tests only on the Windows
| ID | Sev | Tier | Eff | Dep | Status | Title |
|----|-----|------|-----|-----|--------|-------|
| WRK-21 | Medium | P0 | M | IPC-23 (same defect, fix owned here); WRK-28 (same lines) | Done | DrainEvents bound is count-based only; an oversized reply still kills the session and loses the drained events |
| WRK-22 | Low | — | S | IPC-26 (same defect, fix owned here) | Not started | Cancelled `WriteAsync` leaves its frame queued; it is still written later |
| WRK-22 | Low | — | S | IPC-26 (same defect, fix owned here) | Done | Cancelled `WriteAsync` leaves its frame queued; it is still written later |
| WRK-23 | Low | — | S | WRK-21 (rejection path becomes backstop-only) | Done | Rejected frames consume sequence numbers, producing wire gaps |
| WRK-24 | Low | — | S | — | Not started | `AdoptNegotiatedMaxMessageBytes` has no lower-bound sanity check |
| WRK-25 | Low | P2 | S | WRK-22 (both touch enqueue/dequeue) | Not started | WRK-12 flush coalescing never engages on the event hot path |
| WRK-24 | Low | — | S | — | Done | `AdoptNegotiatedMaxMessageBytes` has no lower-bound sanity check |
| WRK-25 | Low | P2 | S | WRK-22 (both touch enqueue/dequeue) | Done | WRK-12 flush coalescing never engages on the event hot path |
| WRK-26 | Low | P1 | S | WRK-23 (soft — sequence prose); discharges IPC-29 | Done | Write-priority and overflow doc drift from the WRK-07 change |
| WRK-27 | Low | — | S | — | Not started | Alarm poll bypasses the watchdog's in-flight suppression (15 s vs 75 s) |
| WRK-27 | Low | — | S | — | Done | Alarm poll bypasses the watchdog's in-flight suppression (15 s vs 75 s) |
| WRK-28 | Low | — | S | WRK-21 (land in the same commit cluster) | Done | 10,000 drain cap is a duplicated magic constant with a comment-only sync contract |
---
@@ -15,8 +15,8 @@ All `path:line` citations were re-verified against the working tree at `4f5371f`
| IPC-23 | Medium | P0 | S¹ | WRK-21 | Done | DrainEvents bound is count-based only; byte-heavy queue still builds a session-killing reply frame (contract requirements here; fix mechanics in WRK-21) |
| IPC-24 | Medium | P0 | S | — | Done | CI's unconditional Java churn-revert masks real generated-code drift for message-level proto changes |
| IPC-25 | Medium | P0 | M | — | Done | Committed Go/Python worker bindings are stale at HEAD; no guard covers them |
| IPC-26 | Low | P2 | S¹ | WRK-22 | Not started | Cancelled write leaves a ghost frame that is still written (contract requirement here; fix mechanics in WRK-22) |
| IPC-27 | Low | P2 | S | — | Not started | Descriptor freshness test blind to enums, enum values, services/methods, and the Galaxy contract |
| IPC-26 | Low | P2 | S¹ | WRK-22 | Done (mechanics landed in WRK-22) | Cancelled write leaves a ghost frame that is still written (contract requirement here; fix mechanics in WRK-22) |
| IPC-27 | Low | P2 | S | — | Done | Descriptor freshness test blind to enums, enum values, services/methods, and the Galaxy contract |
| IPC-28 | Low | — | S | — | Done | `docs/Grpc.md` omits the `CommandTooLarge` → `ResourceExhausted` mapping |
| IPC-29 | Low | — | S | — | Done (discharged by WRK-26) | Worker writer priority scheduling and write-time sequence stamping undocumented in the frame-protocol doc |
| IPC-30 | Low | P0 | M | WRK-21 (same file/batch) | Done | Oversized worker→gateway event frame is session-fatal — make the death deliberate, structured, and diagnosable |
@@ -20,7 +20,7 @@ Operating constraints carried from prior work:
| CLI-36 | Medium | P0 | S | — | Done | Go CLI `stream-events` silently destroys the ReplayGap signal |
| CLI-37 | Medium | P1 | M | CLI-38 | Done | Status-array validation must branch on `category` per the proto contract (4-vs-1 divergence) |
| CLI-38 | Medium | P1 | S | — | Done | Align .NET/Go/Java on `hresult < 0` — lands prior CLI-08 and cures the design-doc drift |
| CLI-39 | Medium | P1 | S | CLI-35..38, CLI-45 | Not started | Bump client versions off the already-published 0.1.2 before the next publish; add registry-collision guard |
| CLI-39 | Medium | P1 | S | CLI-35..38, CLI-45 | Done | Bump client versions off the already-published 0.1.2 before the next publish; add registry-collision guard |
| CLI-40 | Low | — | M | — | Done | Port the exact-secret credential scrub to Rust/Java/.NET |
| CLI-41 | Low | — | M | — | Done | Uniform malformed-reply contract for AuthenticateUser/ArchestrAUserToId/AddBufferedItem |
| CLI-42 | Low | P1 | S | — | Done | Document the vendored Rust proto layout (CLI-02's missing doc half) |
@@ -15,7 +15,7 @@ Prior-cycle open findings (TST-05..24 where still open) are tracked in the prior
| TST-27 | Medium | P1 (doc batch) | S | — | Done | `ShowTagValues` config row still says "Reserved" after SEC-25 made the flag live |
| TST-28 | Low | P2 | S | relates IPC-02 | Done | Gateway-side `max_frame_bytes` handshake field untested in the CI-run suite |
| TST-29 | Low | P2 | S | — | Done | Retire `oldtasks.md` after folding the Phase-5 governance record into DesignDecisions.md; delete root docs-review artifacts |
| TST-30 | Low | P2 | M | — | Not started | Single shared Gitea runner is a CI throughput/availability bottleneck (cross-repo contention, no run cancel/delete) |
| TST-30 | Low | P2 | M | — | Done (doc half; runner registration operator-pending per runbook) | Single shared Gitea runner is a CI throughput/availability bottleneck (cross-repo contention, no run cancel/delete) |
---
@@ -167,6 +167,8 @@ Independent of the runner count, document the **no-cancel** reality (Gitea 1.26
**Verification.** Push two branches back-to-back and confirm their runs execute concurrently (not serially) once a second runner exists; `GET /repos/dohertj2/mxaccessgw/actions/runners` (or the instance runner list) shows ≥2 runners online; `docs/GatewayTesting.md` describes the shared-runner/no-cancel reality and the bypass. Re-run the TST-25 acceptance push and confirm queue depth is materially lower under a concurrent `lmxopcua` run.
**Outcome (2026-08-07 — Done, doc half; runner registration operator-pending).** Landed on `fix/tst-30-runner-docs`. Implementation step 2 shipped: `docs/GatewayTesting.md`'s Continuous Integration section gained a "Runner capacity is shared and finite" subsection stating the `maxParallel=1` co-located runner is shared with `dohertj2/lmxopcua` at the instance level (not repo-scoped), the ~20–30 minute queue latency observed under cross-repo contention, and the Gitea 1.26 no-cancel/no-delete API reality; the existing "windev tier down" degraded-mode paragraph now also covers "runner contended" as a reason to use the bypass, generalized per this finding's design note. New operator runbook `docs/runbooks/TST-30-second-ci-runner.md` carries **step 1** (register a second `act_runner` on `10.100.0.35`, option (a) recommended, same `container.network: traefik` config; option (b) dedicated labelled runner as an escalation; option (c) windev-hosted runner rejected) with the verification checklist (concurrent back-to-back pushes, `GET /repos/dohertj2/mxaccessgw/actions/runners` ≥ 2) and a note that the no-cancel reality persists regardless of runner count. **Step 3 (optional workflow-level `concurrency` group)** is documented in the runbook as unverified — explicitly framed as "verify this Gitea deployment honors it before relying on it" — and left unimplemented in `ci.yml`, since it is a `ci.yml` change out of scope for this doc-only pass. **The actual runner registration (step 1) is infrastructure work outside this repo's tree and remains the operator's to execute**, tracked in the runbook. Verification performed: `grep -n 'maxParallel\|shared\|cancel' docs/GatewayTesting.md` shows the new prose; runbook file exists at the path above; no build required (doc-only change).
---
## Cross-domain dependencies
@@ -0,0 +1,19 @@
# Candidate Findings for the Next Review Cycle (surfaced during 2026-07-12 remediation)
These were discovered while remediating the 2026-07-12 backlog but were **out of scope** for it — each is either pre-existing, by-design residual, or a new observation. They are recorded here (not fixed) so the next review cycle can triage them. None blocks the 2026-07-12 cycle, which is complete.
| ID (proposed) | Area | Severity (est.) | Summary |
|---|---|---|---|
| NEXT-01 | Testing / macOS | Low | Fake-worker/e2e gateway tests fail on macOS under the default `TMPDIR` because the `CoreFxPipe_mxaccess-gateway-{pid}-{sessionId}` path exceeds the 104-char Unix-domain-socket `sun_path` limit under `/var/folders/…/T/`. Workaround today is `TMPDIR=/tmp`. Fix options: shorten the pipe name, or document the `TMPDIR=/tmp` requirement in `docs/GatewayTesting.md`. Surfaced independently by multiple remediation agents. |
| NEXT-02 | Clients (.NET, Java) | Low | The .NET and Java CLIs render the raw `ReplayGap` sentinel `MxEvent` on `stream-events` instead of a typed gap row — Java text mode prints `0 MX_EVENT_FAMILY_UNSPECIFIED`. Same defect class as CLI-36 (Go) / CLI-35 (Python), which were fixed this cycle; the .NET/Java halves were out of scope. The cross-language smoke matrix now records this divergence honestly. |
| NEXT-03 | Gateway alarms | Low | `GatewayAlarmMonitor.ApplyReconcile` feed-repair broadcasts (the new acked-delta from GWC-26 **and** the pre-existing Raise/Clear repair) are **at-least-once, not exactly-once**: a periodic reconcile can synthesize a transition whose matching live transition is still buffered in the alarm lease, so both broadcast as indistinguishable duplicates on the alarm feed (StreamAlarms + dashboard hub). Pre-existing (the Raise/Clear repair always had it); GWC-26 documented the at-least-once contract rather than closing the race. Closing it needs reconcile/live serialization or a monotonic dedup marker. |
| NEXT-04 | Worker frame writer | Low | WRK-22/WRK-25 cancellation path: a frame `Claimed` by a concurrent lock-holder just before its caller's cancellation races in is never awaited by that caller; if the write then faults, `TrySetException` lands on a `Task` nobody observes (unobserved-task-exception). By-design residual, non-crash (no `UnobservedTaskException` handler registered), pre-existing to single-frame WRK-22 and amplified per-batch by WRK-25. Hygiene fix: attach a fault-observing continuation to abandoned/tombstoned frame completions. |
| NEXT-05 | Worker frame writer | Info | A batch whose remaining frames are tombstoned by cancellation leaves dead `PendingFrame` entries in `_eventFrames`/`_controlFrames` until a future `DequeueNext` pops and skips them. Same pre-existing behavior as single-frame WRK-22, amplified per-batch; in practice heartbeats purge them promptly, so not a real leak. |
## Operator actions still pending (from this cycle's runbooks)
These are **live-infrastructure actions the operator must execute** — the repo-side work is complete and merged:
- **SEC-36** — rotate the dev LDAP service-account credential per `docs/runbooks/SEC-36-ldap-credential-rotation.md` (generate new secret in `scadaproj/infra/glauth`, pre-stage the NSSM env var on deployed hosts, rotate GLAuth on `10.100.0.35`, verify dashboard login). The committed literal is gone from the working tree but remains recoverable from git history until rotation completes — **rotation is the load-bearing half.**
- **TST-30** — register a second Gitea `act_runner` on `10.100.0.35` per `docs/runbooks/TST-30-second-ci-runner.md` to relieve the single-shared-runner bottleneck.
- **TST-25 follow-ups** — old **TST-05** (scheduled live-MXAccess smoke) is now covered by the `nightly-windev` job; old **TST-24** (client wire tests in CI) is unblocked by the working Windows tier.
+2 -2
View File
@@ -197,7 +197,7 @@ Full design + implementation for each row lives in the linked domain doc under i
| CLI-21 | Low | P2 | S | — | Done | Go `ClientVersion = "0.1.0-dev"` stale vs tagged releases |
| CLI-22 | Low | — | S | — | Not started | Go `newCorrelationID` swallows `crypto/rand` error → empty id |
| CLI-23 | Low | — | S | — | Not started | Go nil-vs-empty bulk short-circuit asymmetry |
| CLI-24 | Low | — | S | — | Not started | Java `MxEventStream` single-consumer constraint undocumented |
| CLI-24 | Low | — | S | — | Done | Java `MxEventStream` single-consumer constraint undocumented (closed 2026-08-07 per 2026-07-12 review old-tracker action; documented at MxEventStream.java:25 "Single consumer") |
| CLI-25 | Low | — | S | — | Not started | Java `close()` does not await channel termination |
| CLI-26 | Low | P2 | S | — | Done | Python `version.py` (0.1.0) ≠ `pyproject.toml` (0.1.2) |
| CLI-27 | Low | — | S | — | Not started | Python `Session.close()` not concurrency-safe; synthesizes reply |
@@ -207,7 +207,7 @@ Full design + implementation for each row lives in the linked domain doc under i
| CLI-31 | Low | — | M | — | Not started | Rust CLI is a single 2,699-line `main.rs` |
| CLI-32 | Low | — | S | — | Not started | Client-side bulk caps differ (.NET/Java unbounded) |
| CLI-33 | Low | — | S | CLI-01,13 | Not started | Per-language event backpressure semantics undocumented |
| CLI-34 | Low | — | S | — | Not started | Python `build/`/`.pytest_cache/` present on disk (untracked) |
| CLI-34 | Low | — | S | — | Done | Python `build/`/`.pytest_cache/` present on disk (untracked) (closed 2026-08-07 per 2026-07-12 review old-tracker action; both gitignored in clients/python/.gitignore) |
### Testing, docs & gaps — [60-testing-docs-gaps.md](60-testing-docs-gaps.md)
+1 -1
View File
@@ -487,7 +487,7 @@ dotnet nuget add source https://gitea.dohertylan.com/api/packages/dohertj2/nuget
Then add the package to your project:
````bash
dotnet add package ZB.MOM.WW.MxGateway.Client --version 0.1.1
dotnet add package ZB.MOM.WW.MxGateway.Client --version 0.2.0
````
The `ZB.MOM.WW.MxGateway.Contracts` package is pulled in transitively.
@@ -19,7 +19,7 @@
<PropertyGroup>
<IsPackable>true</IsPackable>
<PackageId>ZB.MOM.WW.MxGateway.Client</PackageId>
<Version>0.1.2</Version>
<Version>0.2.0</Version>
<Description>.NET 10 gRPC client for the MxAccessGateway service. Provides typed wrappers, retry, and a lazy-browse walker over the Galaxy Repository hierarchy.</Description>
<PackageReadmeFile>README.md</PackageReadmeFile>
<!-- Only the shipped library generates XML docs (matching src/Contracts). The Cli and
+5 -3
View File
@@ -471,7 +471,7 @@ go run ./cmd/mxgw-go smoke -endpoint $env:MXGATEWAY_ENDPOINT -plaintext -api-key
The module is resolved directly from the git repo — no package registry:
````bash
go get gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go@v0.1.1
go get gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go@v0.2.0
````
Then import:
@@ -494,11 +494,13 @@ Go modules in monorepo subdirectories use prefixed tags. To tag a release
from this repo:
````bash
pwsh scripts/tag-go-module.ps1 -Version v0.1.1 -Push
pwsh scripts/tag-go-module.ps1 -Version v0.2.0 -Push
````
The script validates semver, refuses to tag with uncommitted tracked
changes, creates an annotated tag `clients/go/v0.1.1`, and (with `-Push`)
changes, verifies `clients/go/mxgateway/version.go`'s `ClientVersion`
matches the requested tag version (failing the tag otherwise — CLI-21/CLI-39),
creates an annotated tag `clients/go/v0.2.0`, and (with `-Push`)
pushes it to origin.
## Related Documentation
+1 -1
View File
@@ -3,7 +3,7 @@ package mxgateway
const (
// ClientVersion is the released semantic version of this Go client module.
// Keep it in sync with the module tag applied by scripts/tag-go-module.ps1.
ClientVersion = "0.1.2"
ClientVersion = "0.2.0"
// GatewayProtocolVersion matches GatewayContractInfo.GatewayProtocolVersion
// in the shared .NET contracts.
+1 -1
View File
@@ -465,7 +465,7 @@ repositories {
}
dependencies {
implementation 'com.zb.mom.ww.mxgateway:zb-mom-ww-mxgateway-client:0.1.2'
implementation 'com.zb.mom.ww.mxgateway:zb-mom-ww-mxgateway-client:0.2.1'
}
````
+8 -1
View File
@@ -13,7 +13,14 @@ ext {
subprojects {
group = 'com.zb.mom.ww.mxgateway'
version = '0.2.0'
// 0.2.0 was already published to the Gitea Maven feed on 2026-06-26,
// before the CLI-37/38/40/41 conformance fixes changed the client's
// observable behavior (status.category-based validation, hresult < 0,
// exact-secret redaction, typed malformed-reply errors). Bump to 0.2.1
// so the published coordinate matches the conformant behavior the other
// four clients ship at 0.2.0 for the first time. See CLI-39 and the
// "Versioning" section of docs/ClientPackaging.md.
version = '0.2.1'
pluginManager.withPlugin('java') {
java {
@@ -59,7 +59,7 @@ final class MxGatewayCliTests {
assertEquals(0, run.exitCode());
assertEquals("", run.errors());
assertTrue(run.output().contains("mxgateway-java 0.2.0"));
assertTrue(run.output().contains("mxgateway-java 0.2.1"));
assertTrue(run.output().contains("gatewayProtocolVersion=3"));
assertTrue(run.output().contains("workerProtocolVersion=1"));
}
@@ -89,7 +89,7 @@ final class MxGatewayCliTests {
CliRun run = execute(new FakeClientFactory(), "version", "--json");
assertEquals(0, run.exitCode());
assertTrue(run.output().contains("\"clientVersion\":\"0.2.0\""));
assertTrue(run.output().contains("\"clientVersion\":\"0.2.1\""));
assertTrue(run.output().contains("\"gatewayProtocolVersion\":3"));
}
@@ -9,7 +9,7 @@ package com.zb.mom.ww.mxgateway.client;
public final class MxGatewayClientVersion {
private static final int GATEWAY_PROTOCOL_VERSION = 3;
private static final int WORKER_PROTOCOL_VERSION = 1;
private static final String CLIENT_VERSION = "0.2.0";
private static final String CLIENT_VERSION = "0.2.1";
private MxGatewayClientVersion() {
}
+1 -1
View File
@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "zb-mom-ww-mxaccess-gateway-client"
version = "0.1.2"
version = "0.2.0"
description = "Async Python client for MXAccess Gateway."
readme = "README.md"
requires-python = ">=3.12"
@@ -1,3 +1,3 @@
"""Package version information."""
__version__ = "0.1.2"
__version__ = "0.2.0"
+17
View File
@@ -1,6 +1,8 @@
"""Tests for the Python CLI."""
import json
import tomllib
from pathlib import Path
import pytest
from click.testing import CliRunner
@@ -12,6 +14,21 @@ from zb_mom_ww_mxgateway_cli.commands import main
_BATCH_EOR = "__MXGW_BATCH_EOR__"
def test_version_matches_pyproject_toml() -> None:
"""`__version__` must track `pyproject.toml`'s `[project].version`.
The existing `version` command tests only assert self-consistency against
`__version__` (the two hardcoded literals could still drift from each
other without either test catching it — the CLI-26 residual drift mode).
This test pins `__version__` to the single source of truth instead.
"""
pyproject_path = Path(__file__).resolve().parent.parent / "pyproject.toml"
with pyproject_path.open("rb") as handle:
pyproject = tomllib.load(handle)
assert __version__ == pyproject["project"]["version"]
def test_require_certificate_validation_flag_flows_through_connect(
monkeypatch: pytest.MonkeyPatch,
) -> None:
+2 -2
View File
@@ -590,7 +590,7 @@ checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
[[package]]
name = "mxgw-cli"
version = "0.1.2"
version = "0.2.0"
dependencies = [
"clap",
"futures-util",
@@ -1490,7 +1490,7 @@ dependencies = [
[[package]]
name = "zb-mom-ww-mxgateway-client"
version = "0.1.2"
version = "0.2.0"
dependencies = [
"futures-core",
"futures-util",
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "zb-mom-ww-mxgateway-client"
version = "0.1.2"
version = "0.2.0"
edition = "2021"
authors = ["Joseph Doherty"]
description = "Async Rust client for the MxAccessGateway gRPC service, including a lazy-browse walker over the Galaxy Repository hierarchy."
@@ -25,7 +25,7 @@ resolver = "2"
[workspace.package]
edition = "2021"
version = "0.1.2"
version = "0.2.0"
authors = ["Joseph Doherty"]
license = "Proprietary"
repository = "https://gitea.dohertylan.com/dohertj2/mxaccessgw"
+1 -1
View File
@@ -436,5 +436,5 @@ Then add the dependency:
```toml
[dependencies]
zb-mom-ww-mxgateway-client = { version = "0.1.1", registry = "dohertj2-gitea" }
zb-mom-ww-mxgateway-client = { version = "0.2.0", registry = "dohertj2-gitea" }
```
+67
View File
@@ -32,6 +32,73 @@ $env:MXGATEWAY_TEST_ITEM = 'TestObject.TestInt'
Use plaintext only for a local gateway. Use TLS when the gateway crosses a
machine boundary or uses a production certificate.
## Versioning
Every client's version lives in its own manifest: `clients/rust/Cargo.toml`
(`[package]` and `[workspace.package]`, both must match — `crates/mxgw-cli`
inherits via `version.workspace = true`), `clients/python/pyproject.toml`
(`[project].version`) and `clients/python/src/zb_mom_ww_mxgateway/version.py`
(`__version__`, must match `pyproject.toml`), `clients/go/mxgateway/version.go`
(`ClientVersion`), `clients/dotnet/ZB.MOM.WW.MxGateway.Client/ZB.MOM.WW.MxGateway.Client.csproj`
(`<Version>`), and `clients/java/build.gradle` (`subprojects { version = ... }`,
mirrored by the hand-maintained `MxGatewayClientVersion.CLIENT_VERSION`
constant — the two have drifted before and there is no build-time link
between them, so bump both together).
`src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj`
(`<Version>`) is a fifth, easy-to-miss manifest: it is not itself a
language client, but `Invoke-PackDotnet` in `scripts/pack-clients.ps1`
packs and publishes it in lockstep with the .NET Client (both
`ZB.MOM.WW.MxGateway.*` nupkgs go through the same `-Publish` loop), and
Contracts and the .NET Client have always released at the same version.
Bump Contracts' `<Version>` alongside the .NET Client's — leaving it behind
means the next `-Publish` packs a stale Contracts version, the collision
guard below correctly refuses to re-publish it, and the loop aborts
mid-way with the Client possibly already pushed (nupkgs are enumerated
alphabetically, and `Client` sorts before `Contracts`). This is distinct
from `src/Directory.Build.props`'s repo-wide `<Version>` default, which
stamps the Server/Worker/test assemblies and is not part of the published
client package set — see the comment there.
**Bump the version before every publish, never after.** A Gitea package feed
rejects re-uploading an existing name+version, and `scripts/pack-clients.ps1`
enforces this before it ever attempts a push: each per-language `-Publish`
step queries the Gitea package API
(`GET /api/v1/packages/dohertj2/{type}/{name}/{version}`) for the version
about to be published and aborts with a clear error if it already exists —
the script never force-overwrites a published artifact. `scripts/tag-go-module.ps1`
carries the equivalent guard for the Go module: it refuses to create a
`clients/go/vX.Y.Z` tag unless `clients/go/mxgateway/version.go`'s
`ClientVersion` already equals `X.Y.Z` (CLI-21/CLI-39), so a forgotten
version bump fails the tag instead of shipping a mismatched module.
As of 2026-08-07 (CLI-39) all five clients — plus `ZB.MOM.WW.MxGateway.Contracts`,
which releases in lockstep with the .NET Client — moved to **0.2.0**,
converging on one number after four of the five had drifted onto the
*already-published* 0.1.2/0.1.1 while their public APIs kept changing
underneath it (see `archreview/2026-07-12/remediation/50-clients.md` CLI-39).
A code-review follow-up on the same branch caught that the initial CLI-39
pass bumped the .NET Client but left `Contracts.csproj` at 0.1.2 — since
both publish through the same `Invoke-PackDotnet` `-Publish` loop, that
would have made the very next `.NET` publish abort on the new collision
guard partway through (Client already pushed, Contracts refused as a
re-publish of the already-published 0.1.2). Fixed in the same branch.
Verified against the live Gitea package API at that time: `nuget` had
`ZB.MOM.WW.MxGateway.Client` and `.Contracts` published through 0.1.2; `pypi` (`zb-mom-ww-mxaccess-gateway-client`)
and `cargo` (`zb-mom-ww-mxgateway-client`) had only reached 0.1.1 despite their
source pinning 0.1.2; **`maven`
(`com.zb.mom.ww.mxgateway:zb-mom-ww-mxgateway-client`) had already published
0.2.0 on 2026-06-26** — before the CLI-37/38/40/41 conformance fixes changed
the client's observable behavior (`category`-based status validation,
`hresult < 0`, exact-secret redaction, typed malformed-reply errors). Reusing
0.2.0 for the conformant Java build would have labeled two different APIs
with the same coordinate, so **Java is the one exception: it shipped as
0.2.1**, not 0.2.0. Operators publishing a future release must re-check the
target version against the live registry before assuming any of these
numbers are still unclaimed — the guards above do this automatically at
publish time, but a version bump in the source is still a manual step per
client.
## .NET
The .NET client uses .NET 10 and references
+6 -4
View File
@@ -74,10 +74,12 @@ protoc-version encoding drift and does not false-fail across protoc releases
(it warns, rather than fails, when protoc is off the pin).
The gateway test project carries an independent, protoc-free freshness guard:
`ClientProtoInputTests.Descriptor_ContainsEveryContractMessageAndField` reflects
over the in-process contract descriptors and fails if any contract message or
field is missing from the committed protoset. This is the primary CI gate for
descriptor staleness; a red test means "regenerate and commit the protoset."
`ClientProtoInputTests.Descriptor_ContainsEveryContractSymbol` reflects over the
in-process contract descriptors — `mxaccess_gateway.proto`, `mxaccess_worker.proto`,
and `galaxy_repository.proto` — and fails if any contract message, field, enum,
enum value, service, or method is missing from the committed protoset. This is
the primary CI gate for descriptor staleness; a red test means "regenerate and
commit the protoset."
### Pinned generator versions
+5 -4
View File
@@ -156,10 +156,11 @@ pwsh -File scripts/publish-client-proto-inputs.ps1
Freshness is guarded two ways so a skipped regeneration cannot ship silently:
- `ClientProtoInputTests.Descriptor_ContainsEveryContractMessageAndField` (gateway test project)
reflects over the in-process contract descriptors and fails if any message or field is missing
from the committed protoset. It is semantic (symbol presence), needs no protoc, and runs in the
Linux CI. A red test means "regenerate and commit the protoset."
- `ClientProtoInputTests.Descriptor_ContainsEveryContractSymbol` (gateway test project)
reflects over the in-process contract descriptors — including `galaxy_repository.proto` — and
fails if any message, field, enum, enum value, service, or method is missing from the committed
protoset. It is semantic (symbol presence), needs no protoc, and runs in the Linux CI. A red
test means "regenerate and commit the protoset."
- `pwsh -File scripts/publish-client-proto-inputs.ps1 -Check` rebuilds the descriptor and compares
it to the committed one. The comparison normalizes both sides through the same protoc with
`source_code_info` stripped, so it does not false-fail across protoc releases.
+33 -5
View File
@@ -453,6 +453,32 @@ runtime because the x86 Worker cannot build on Linux:
`xUnit1030`). Gated `if: github.event_name == 'schedule'`, so it never gates a push. On
failure it opens a Gitea issue via the Actions token, since nobody watches the Actions page.
### Runner capacity is shared and finite
All CI runs on a co-located `gitea-runner` container on docker host `10.100.0.35` with
`maxParallel=1`, and that runner instance is **shared across repos** — it interleaves
`dohertj2/mxaccessgw` and `dohertj2/lmxopcua` jobs on the single slot rather than being
scoped to this repo (`GET /repos/dohertj2/mxaccessgw/actions/runners` returns
`total_count: 0`; the runner is registered at the instance level). Every job in a run
(`portable`, `java`, `windows-x86`) executes serially, so queue latency is additive within
a run and an active `lmxopcua` run blocks `mxaccessgw` entirely — expect ~20–30 minutes of
queue depth under cross-repo contention, not a stuck pipeline. This Gitea version (1.26)
also exposes **no run cancel or delete via the API** (`POST .../actions/runs/{id}/cancel`
returns 404, `DELETE .../actions/runs/{id}` returns 400), so a superseded or hung run cannot
be cleared and holds the slot until it finishes or times out. See
`docs/runbooks/TST-30-second-ci-runner.md` for the operator runbook that registers a second
runner to relieve this; until that lands, treat single-runner contention as expected, not a
CI outage.
When queue depth (or the missing-cancel reality) makes waiting impractical, verify a
specific commit out of band instead of waiting behind the queue: run
`CI_SHA=<sha> scripts/ci/run-windev-ci.sh <build|test|live>` from a machine with SSH access
to windev (the same script the SSH-driven `windows-x86`/`nightly-windev` jobs use — see
`scripts/ci/README.md`), or fall back to the manual windev worktree procedure below. This is
the same escape hatch used when the windev tier itself is down — TST-30 generalizes it from
"tier down" to "runner contended": either way, a stuck or slow shared runner should not
block verifying a commit.
The freshness guard `scripts/check-codegen.ps1` runs four checks and fails the build when the
committed client descriptor set (Check 1), the C# `Generated/` (Check 2), the Rust vendored
protos (Check 3), or the Go/Python client bindings (Check 4, IPC-25) no longer match the current
@@ -467,11 +493,13 @@ the x86 build with `CS0246`). See [Client Proto Generation](./ClientProtoGenerat
[Contracts](./Contracts.md).
If the SSH-driven Windows tier is unavailable for infrastructure reasons (windev down, CI
key/secret rotation in flight), fall back to the manual windev worktree procedure as a
degraded mode: on windev, fast-forward an isolated `origin/main` worktree under `C:\build`
(never the dirty Desktop checkout), then run the x86 Worker build and `Worker.Tests`
(`-p:Platform=x86`) there by hand. Do this per merge for worker-touching changes until the
`windows-x86` job is green again.
key/secret rotation in flight) **or** the shared Gitea runner is contended and the queue is
impractical to wait behind (see "Runner capacity is shared and finite" above), fall back to
the manual windev worktree procedure as a degraded mode: on windev, fast-forward an isolated
`origin/main` worktree under `C:\build` (never the dirty Desktop checkout), then run the x86
Worker build and `Worker.Tests` (`-p:Platform=x86`) there by hand. Do this per merge for
worker-touching changes until the `windows-x86` job is green again (tier-down case) or the
queue clears (contention case).
## Related Documentation
+17
View File
@@ -747,6 +747,23 @@ heartbeat fields until dedicated thresholds own those warnings. The worker
reports stale STA activity, but the gateway owns the final kill decision
through its existing heartbeat and worker lifecycle policy.
The alarm poll runs outside the command dispatcher — `RunAlarmPollLoopAsync`
invokes `PollOnce` directly on the STA rather than through
`StaCommandDispatcher`, so it does not inflate `PendingCommandCount` or perturb
dispatch ordering for real gateway commands. Because it is not a dispatched
command it has no `CurrentCommandCorrelationId`, so a healthy-but-slow poll (a
large `GetXmlCurrentAlarms2` against a busy provider) blocking the STA past
`HeartbeatGrace` would otherwise fault a healthy session at 15 s while a
dispatched command gets the 75 s ceiling. To close that asymmetry (WRK-27) the
poll advertises itself on the heartbeat snapshot's `StaCallInProgress` flag —
set on the STA thread for exactly the span of the COM call — and the watchdog
suppression honors that flag alongside `CurrentCommandCorrelationId`. The poll
therefore receives the same grace-to-ceiling treatment as a dispatched command:
suppressed up to `HeartbeatStuckCeiling`, faulted past it (a poll that blocks
the STA more than 75 s without pumping *should* fault — that is the ceiling's
contract). The flag is named generically so any future non-dispatcher STA work
reuses it.
The in-flight-command suppression itself is bounded by
`WorkerPipeSessionOptions.HeartbeatStuckCeiling` (default 75 seconds = 5 ×
`HeartbeatGrace`). The motivating case for the suppression is a legitimately
+33 -13
View File
@@ -29,6 +29,17 @@ default. A `max_frame_bytes` of 0 (an older gateway that never set the field)
means "use the worker's built-in default". This keeps both ends framing to the
same limit rather than depending on matched compile-time constants.
The worker accepts a negotiated value in the closed range [1024, 256 MiB]
(`MinNegotiableFrameBytes` .. `MaxNegotiableFrameBytes`); 0 keeps the default.
A value outside that range is rejected at the handshake with a fault frame
rather than adopted, because a nonsensical maximum — a gateway bug or a
foreign/old peer — would otherwise leave a session that handshakes cleanly and
then fails every subsequent frame with per-frame size errors, the worst
diagnostic shape for an operator. The 1024-byte floor matches the gateway's own
`GatewayOptionsValidator.MinimumMaxMessageBytes`, so the worker never rejects a
value the gateway's validator accepts as legal configuration, and 1024 still
guarantees hellos, heartbeats, acks, and faults fit.
Every worker-to-gateway frame must serialize within this limit, control replies
included, so reply builders truncate to fit rather than emit a frame the writer
will reject. `WorkerPipeSession` pre-sizes a `DrainEvents` reply below the
@@ -112,20 +123,29 @@ runs after the whole batch, and only then does every successfully-written
frame's completion resolve — so a caller's `WriteAsync` still does not
complete until its bytes are both written *and* flushed, but a batch that
happened to contain several queued frames pays one flush instead of one per
frame. In practice this coalescing currently engages only when multiple
frames are queued at the moment a lock-holder starts draining. The event
drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`) awaits each drained
event's `WriteAsync` individually before writing the next, so today at most
one event frame is queued per drain pass and each event still costs its own
flush; a dedicated batch write entry point that submits a whole drained
event batch under one lock acquisition is designed but not yet landed, so a
burst of N events currently costs N flushes on the event hot path, not one.
frame. The event drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`)
submits a whole drained event batch through `WriteBatchAsync`, which enqueues
every frame under one `_gate` acquisition, takes the write lock once, and
drains them together, so a burst of N events costs one flush rather than N —
the coalescing the batch machinery was built for now engages on the event hot
path, not only when independent producers happen to queue behind a blocked
write. Intra-batch order is preserved (FIFO enqueue under one lock), and a
concurrently queued control frame is still drained ahead of the batch. A
per-frame rejection inside a batch (for example one oversized event) surfaces
from the batch's awaited completions as that frame's
`WorkerFrameProtocolException`; the remaining completions are still observed
so none faults unobserved.
Cancellation semantics for a `WriteAsync` call that is still waiting for the
write lock when its token fires are not yet defined at this layer — pending
a fix that will tombstone the queued frame so a cancelled call is guaranteed
never to reach the wire. Until that lands, a cancelled caller may still see
its frame written by whichever caller next holds the lock.
Cancellation of a `WriteAsync`/`WriteBatchAsync` call that is still waiting
for the write lock when its token fires tombstones the queued frame: the
cancelled caller marks its frame under `_gate`, and the draining lock-holder's
`DequeueNext` skips any tombstoned frame, so a cancelled call is guaranteed
never to reach the wire — *unless* a lock-holder has already claimed the frame
to write it. Claiming and cancelling are interlocked under `_gate`, so exactly
one wins; a frame already claimed is mid-write and can no longer be recalled,
so the caller observes `OperationCanceledException` while that one frame still
reaches the wire. That residual window is by design: blocking the canceller
behind the very write it is abandoning would defeat the point of cancellation.
## Verification
+113
View File
@@ -0,0 +1,113 @@
# TST-30 — Register A Second CI Runner (Operator Runbook)
Operator steps to relieve the single shared Gitea Actions runner that CI depends on. The
repo-side half of TST-30 (documenting the shared-runner/no-cancel reality and the
`run-windev-ci.sh` bypass) is already landed in `docs/GatewayTesting.md`; registering the
second runner below is infrastructure work outside this repo's tree and is yours to execute.
## Why
All CI for this repo runs on one co-located `gitea-runner` container on docker host
`10.100.0.35` with `maxParallel=1`. That runner is registered at the **instance** level, not
scoped to this repo (`GET /repos/dohertj2/mxaccessgw/actions/runners` returns
`total_count: 0`), so it is shared with `dohertj2/lmxopcua` and every job in every run across
both repos executes serially on the single slot. A `mxaccessgw` push fans out to `portable`,
`java`, `windows-x86`, and an active `lmxopcua` run blocks all of them — queue depth of
~20–30 minutes was observed during TST-25 acceptance under cross-repo contention. Gitea 1.26
also exposes **no run cancel or delete via the API**
(`POST .../actions/runs/{id}/cancel` → 404, `DELETE .../actions/runs/{id}` → 400), so a
superseded or hung run cannot be cleared and holds the slot until it finishes or times out.
This is not a correctness problem — every job still reports accurately — but it undercuts the
fast-feedback purpose of the TST-25 Windows tier and makes CI fragile to a single host: if
`10.100.0.35` wedges or goes down, CI for both repos stops with no failover.
## Options (cheapest first)
- **(a) Register a second `act_runner` instance on `10.100.0.35` — recommended.** The host
already runs `gitea-runner`; add a second `act_runner` container (or raise the existing
runner's `maxParallel` where the docker-in-docker/resource budget allows) so at least two
jobs run concurrently. Cheapest change, and it keeps the runner co-located on the
`container.network: traefik` network that resolves `gitea:3000` — the property TST-03
depended on. **Use the same `container.network: traefik` config as the existing runner.**
- **(b) Dedicate a labelled runner to `mxaccessgw`.** Cleaner isolation — `lmxopcua` load
never blocks this repo — but needs label wiring: register the new runner with a distinct
label (e.g. `mxgw`) and change `.gitea/workflows/ci.yml`'s `runs-on:` for this repo's jobs
to gate on that label (e.g. `runs-on: [ubuntu-latest, mxgw]`). Only do this if (a) proves
insufficient — it is more moving parts for the same throughput gain, and it means `ci.yml`
changes, which is out of scope for the doc-only half of TST-30.
- **(c) Put the runner on windev / a second host — rejected as the primary fix.** windev is
the Windows build target (`10.100.0.48`), not a CI host, and co-locating a Linux runner
there loses the `gitea:3000` name resolution TST-03 relies on. Only consider if
`10.100.0.35` genuinely runs out of capacity for a second instance.
Default to **(a)**. Escalate to (b) only if `lmxopcua` contention persists after a second
instance is online (i.e., (a) is not sufficient because the two repos' combined load exceeds
two slots).
## Preconditions
- SSH/docker access to `10.100.0.35`.
- The existing `gitea-runner` container's compose/run config, to copy its
`container.network: traefik` setting and registration token flow (repo memory
`project_gitea_ci` records this configuration).
- Admin access to Gitea (`gitea.dohertylan.com`) to mint a new runner registration token.
## Steps — option (a): second runner instance
1. On `10.100.0.35`, locate the existing `gitea-runner` container/compose definition and copy
its configuration for a new instance (same `container.network: traefik`, same Docker
socket mount if it uses docker-in-docker, a distinct container name/data volume).
2. In Gitea, generate a new runner registration token (instance-level, since the existing
runner is registered at the instance level too — Admin → Actions → Runners, or
`POST /admin/actions/runners/registration-token`).
3. Register and start the second `act_runner` instance with that token, pointed at the same
Gitea origin.
4. Confirm both runners show online: instance runner list in the Gitea admin UI, or the
equivalent API listing.
## Verification
- Push two branches to `mxaccessgw` back-to-back (or trigger one `mxaccessgw` push while an
`lmxopcua` run is in flight) and confirm both runs execute **concurrently**, not serially —
the second run's jobs should start before the first finishes, not queue behind it.
- `GET /repos/dohertj2/mxaccessgw/actions/runners` (or the instance runner listing) shows
**≥2** runners online.
- Re-run the TST-25 acceptance push (a plain push to a scratch branch) and confirm queue depth
is materially lower than the ~20–30 minute baseline observed under a concurrent `lmxopcua`
run.
- Confirm `windows-x86` still resolves `gitea:3000` correctly from a job scheduled on the new
runner instance (the `traefik` network property must hold for both instances).
## The no-cancel reality does not go away
A second runner relieves contention; it does not add a cancel/delete API — Gitea 1.26 still
returns 404/400 for both. A stale or hung run on either runner still holds its slot until it
finishes or times out. Two runners just means one stale run blocks at most half the capacity
instead of all of it. Do not treat the second runner as a substitute for the escape hatch: a
specific commit can still be verified out of band without waiting on either runner via
`CI_SHA=<sha> scripts/ci/run-windev-ci.sh <build|test|live>` (Linux, needs SSH access to
windev) or the manual windev worktree flow — see the "Runner capacity is shared and finite"
section in `docs/GatewayTesting.md`.
## Optional: workflow-level `concurrency` group
As belt-and-suspenders against the missing cancel API, `.gitea/workflows/ci.yml` could add a
top-level `concurrency` group (e.g. keyed on `${{ github.ref }}`) so a newer push to the same
branch automatically supersedes an in-flight run instead of both running to completion.
**Verify this Gitea deployment actually honors `concurrency` and cancels the superseded run
before relying on it** — Gitea Actions' YAML surface does not track GitHub Actions feature
parity release-for-release, and a `concurrency` block that is silently ignored would look like
a working safeguard while doing nothing. If verified working, this is a `ci.yml` change (not
covered by this runbook) and should land as its own small change with its own verification
(push twice to the same branch quickly, confirm the first run's jobs cancel).
## Done criteria
- A second `act_runner` instance (or raised `maxParallel`) is online on `10.100.0.35` with the
same `container.network: traefik` configuration as the existing runner.
- `GET /repos/dohertj2/mxaccessgw/actions/runners` (or the instance listing) shows ≥2 runners.
- Two concurrent runs (one `mxaccessgw`, one `lmxopcua`, or two `mxaccessgw` pushes) execute
in parallel rather than serially.
- `docs/GatewayTesting.md`'s shared-runner/no-cancel prose and the `run-windev-ci.sh` bypass
remain accurate (they describe the bypass as still valid, which it is regardless of runner
count).
+116
View File
@@ -87,6 +87,12 @@ $GiteaNugetFeed = 'https://gitea.dohertylan.com/api/packages/dohertj2/nuget/inde
$GiteaPypiFeed = 'https://gitea.dohertylan.com/api/packages/dohertj2/pypi'
$JavaHome = '/Users/dohertj2/.local/jdks/jdk-21.0.11+10/Contents/Home'
# Generic Gitea package registry API (https://gitea.dohertylan.com/api/v1/packages/{owner}/{type}/{name}/{version}):
# returns 200 when that exact name+version already exists in the given feed
# type, 404 when it does not. Used as a pre-publish collision guard (CLI-39)
# so a re-run of this script can never silently overwrite a published artifact.
$GiteaPackageApiBase = 'https://gitea.dohertylan.com/api/v1/packages/dohertj2'
function Write-Header {
param([string]$Text)
Write-Host ''
@@ -94,6 +100,64 @@ function Write-Header {
Write-Host $Text -ForegroundColor Cyan
}
function Test-GiteaPackageExists {
<#
.SYNOPSIS
Queries the Gitea package API for an existing name+version in a feed.
.OUTPUTS
$true if the package/version already exists, $false if it does not.
Throws if the registry cannot be reached or returns anything other
than 200/404 — callers must treat "cannot verify" as "do not publish".
#>
param(
[Parameter(Mandatory)][string]$Type,
[Parameter(Mandatory)][string]$Name,
[Parameter(Mandatory)][string]$Version
)
$uri = "$GiteaPackageApiBase/$Type/$Name/$Version"
$headers = @{}
if (-not [string]::IsNullOrEmpty($env:GITEA_TOKEN)) {
$user = if ([string]::IsNullOrEmpty($env:GITEA_USERNAME)) { 'dohertj2' } else { $env:GITEA_USERNAME }
$pair = "$($user):$($env:GITEA_TOKEN)"
$basic = [Convert]::ToBase64String([System.Text.Encoding]::ASCII.GetBytes($pair))
$headers['Authorization'] = "Basic $basic"
}
try {
$response = Invoke-WebRequest -Uri $uri -Headers $headers -Method Get -UseBasicParsing -ErrorAction Stop
return ($response.StatusCode -eq 200)
} catch {
$statusCode = $null
if ($_.Exception.PSObject.Properties['Response'] -and $_.Exception.Response) {
$statusCode = [int]$_.Exception.Response.StatusCode
}
if ($statusCode -eq 404) {
return $false
}
throw "Unable to query the Gitea package API for '$Type/$Name/$Version' ($uri): $($_.Exception.Message). Refusing to publish without a collision check — verify manually (or check GITEA_USERNAME/GITEA_TOKEN/network) and retry."
}
}
function Assert-GiteaPackageNotPublished {
<#
.SYNOPSIS
Aborts the script if $Name/$Version already exists in the $Type feed.
Never force-overwrites a published artifact (CLI-39).
#>
param(
[Parameter(Mandatory)][string]$Type,
[Parameter(Mandatory)][string]$Name,
[Parameter(Mandatory)][string]$Version
)
Write-Host "Checking Gitea '$Type' feed for existing '$Name' $Version..."
if (Test-GiteaPackageExists -Type $Type -Name $Name -Version $Version) {
throw "Gitea package '$Name' version '$Version' already exists in the '$Type' feed. Bump the client version before publishing — this script never force-overwrites a published artifact."
}
Write-Host " Not found in the '$Type' feed — safe to publish '$Name' $Version." -ForegroundColor Green
}
# -------- .NET --------
function Invoke-PackDotnet {
@@ -121,6 +185,15 @@ function Invoke-PackDotnet {
if ($Publish) {
Write-Host 'Publishing .NET packages to Gitea...' -ForegroundColor Yellow
Get-ChildItem $OutputDir -Filter 'ZB.MOM.WW.MxGateway.*.nupkg' | ForEach-Object {
# nupkg filenames are '<PackageId>.<Version>.nupkg'; the id itself contains
# dots (e.g. 'ZB.MOM.WW.MxGateway.Client'), so the id capture is lazy and the
# version capture anchors on the leading digit to split at the right dot.
$fileBaseName = [System.IO.Path]::GetFileNameWithoutExtension($_.Name)
if ($fileBaseName -notmatch '^(?<id>.+?)\.(?<version>\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?)$') {
throw "Could not parse a NuGet package id/version out of '$($_.Name)'."
}
Assert-GiteaPackageNotPublished -Type 'nuget' -Name $Matches.id -Version $Matches.version
& dotnet nuget push $_.FullName --source $GiteaNugetFeed --api-key $env:GITEA_TOKEN
if ($LASTEXITCODE -ne 0) { throw "dotnet nuget push failed for '$($_.Name)'." }
}
@@ -159,6 +232,20 @@ function Invoke-PackPython {
Write-Host "Packed Python artifacts -> $OutputDir" -ForegroundColor Green
if ($Publish) {
$pyprojectPath = Join-Path $RepoRoot 'clients/python/pyproject.toml'
$pyprojectContent = Get-Content $pyprojectPath -Raw
# Scope to the [project] section (not just the first "version = ..." line
# in the file) — [build-system]/[tool.*] sections can carry their own
# version-shaped keys, and matching the file's first hit would be luck
# of ordering, not correctness.
if ($pyprojectContent -notmatch '(?ms)^\[project\](?<section>.*?)(?=^\[|\z)') {
throw "Could not find a [project] section in '$pyprojectPath'."
}
if ($Matches.section -notmatch '(?m)^\s*version\s*=\s*"([^"]+)"') {
throw "Could not find [project].version in '$pyprojectPath'."
}
Assert-GiteaPackageNotPublished -Type 'pypi' -Name 'zb-mom-ww-mxaccess-gateway-client' -Version $Matches[1]
Write-Host 'Publishing Python distribution to Gitea...' -ForegroundColor Yellow
$wheels = @(Get-ChildItem $OutputDir -Filter 'zb_mom_ww_mxaccess_gateway_client-*.whl')
$sdists = @(Get-ChildItem $OutputDir -Filter 'zb_mom_ww_mxaccess_gateway_client-*.tar.gz')
@@ -206,6 +293,20 @@ function Invoke-PackRust {
Write-Host "Packed Rust artifacts -> $OutputDir" -ForegroundColor Green
if ($Publish) {
$cargoTomlPath = Join-Path $rustDir 'Cargo.toml'
$cargoTomlContent = Get-Content $cargoTomlPath -Raw
# Scope to the [package] section specifically — Cargo.toml also carries a
# [workspace.package] section with its own "version = ..." line (today
# identical, by convention, not by anything this regex can rely on), and
# matching whichever comes first in the file is luck of ordering.
if ($cargoTomlContent -notmatch '(?ms)^\[package\](?<section>.*?)(?=^\[|\z)') {
throw "Could not find a [package] section in '$cargoTomlPath'."
}
if ($Matches.section -notmatch '(?m)^\s*version\s*=\s*"([^"]+)"') {
throw "Could not find [package] version in '$cargoTomlPath'."
}
Assert-GiteaPackageNotPublished -Type 'cargo' -Name 'zb-mom-ww-mxgateway-client' -Version $Matches[1]
Write-Host 'Publishing Rust crate to Gitea...' -ForegroundColor Yellow
Push-Location (Join-Path $RepoRoot 'clients/rust')
try {
@@ -269,6 +370,21 @@ function Invoke-PackJava {
Write-Host "Packed Java artifacts -> $OutputDir" -ForegroundColor Green
if ($Publish) {
$buildGradlePath = Join-Path $javaDir 'build.gradle'
$buildGradleContent = Get-Content $buildGradlePath -Raw
if ($buildGradleContent -notmatch "(?m)^\s*group\s*=\s*'([^']+)'") {
throw "Could not find subprojects { group = '...' } in '$buildGradlePath'."
}
$javaGroup = $Matches[1]
if ($buildGradleContent -notmatch "(?m)^\s*version\s*=\s*'([^']+)'") {
throw "Could not find subprojects { version = '...' } in '$buildGradlePath'."
}
$javaVersion = $Matches[1]
# Gitea's Maven package API identifies the package as "groupId:artifactId",
# not the bare artifact id — passing just the artifact id here would query
# a name that never exists and silently defeat the guard.
Assert-GiteaPackageNotPublished -Type 'maven' -Name "$javaGroup`:zb-mom-ww-mxgateway-client" -Version $javaVersion
Write-Host 'Publishing Java artifacts to Gitea Maven feed...' -ForegroundColor Yellow
Push-Location $javaDir
try {
+17
View File
@@ -36,6 +36,23 @@ if ($Version -notmatch '^v\d+\.\d+\.\d+(-[A-Za-z0-9.-]+)?$') {
throw "Version '$Version' must match semver vX.Y.Z (optionally with -prerelease suffix)."
}
# CLI-21 guard: the tag must match what the module itself reports via
# ClientVersion, or `go get <module>@vX.Y.Z` resolves a tag whose module
# code disagrees with its own version constant.
$versionGoPath = Join-Path $PSScriptRoot '..' 'clients/go/mxgateway/version.go'
if (-not (Test-Path $versionGoPath)) {
throw "Could not find '$versionGoPath' to verify ClientVersion before tagging."
}
$versionGoContent = Get-Content $versionGoPath -Raw
if ($versionGoContent -notmatch 'ClientVersion\s*=\s*"([^"]+)"') {
throw "Could not find a ClientVersion = `"...`" constant in '$versionGoPath'."
}
$clientVersion = $Matches[1]
$tagVersion = $Version.TrimStart('v')
if ($clientVersion -ne $tagVersion) {
throw "clients/go/mxgateway/version.go ClientVersion is '$clientVersion' but the requested tag is '$tagVersion'. Update ClientVersion to match before tagging."
}
$tag = "clients/go/$Version"
Write-Host "Creating Go-module tag: $tag" -ForegroundColor Cyan
+8 -4
View File
@@ -11,10 +11,14 @@
<!-- TST-11: single-source the .NET-side version for Server, Worker, Contracts, and tests
(they otherwise stamp the SDK default 1.0.0, so a deployed gateway cannot be correlated
to a release). Kept at 0.1.2 to match the Contracts package and the aligned Python/Rust/
Go clients; the Java client leads at 0.2.0 after its JDK-17 retarget. The git short SHA is
appended to InformationalVersion (0.1.2+<sha>) so support can map a running binary to a
commit; the query is guarded so a build outside a git checkout still succeeds. -->
to a release). Server/Worker/Tests stay at this default. CLI-39 (2026-08-07) moved the
published `ZB.MOM.WW.MxGateway.Contracts` and `.Client` nuget packages to 0.2.0 via an
explicit <Version> override in Contracts.csproj (MSBuild property-last-write-wins over
this Directory.Build.props default) — Server/Worker assembly stamping and the published
client packages are deliberately decoupled; a broader 0.2.0 alignment for Server/Worker
is a separate, not-yet-made decision. The git short SHA is appended to
InformationalVersion (0.1.2+<sha>) so support can map a running binary to a commit; the
query is guarded so a build outside a git checkout still succeeds. -->
<PropertyGroup>
<Version>0.1.2</Version>
</PropertyGroup>
@@ -7,7 +7,7 @@
<PropertyGroup>
<IsPackable>true</IsPackable>
<PackageId>ZB.MOM.WW.MxGateway.Contracts</PackageId>
<Version>0.1.2</Version>
<Version>0.2.0</Version>
<Authors>Joseph Doherty</Authors>
<Company>ZB MOM WW</Company>
<Copyright>Copyright (c) ZB MOM WW. All rights reserved.</Copyright>
@@ -72,18 +72,33 @@ public sealed class ClientBehaviorFixtureTests
foreach (JsonElement fixture in fixtures)
{
string fixtureId = GetFixtureId(fixture);
MxCommandReply reply = ParseFixture<MxCommandReply>(
fixture,
MxCommandReply.Parser);
// Universal invariants: every command-reply fixture parses to a concrete
// command kind and a concrete protocol status, regardless of what MXAccess
// reply detail (if any) it carries.
Assert.NotEqual(MxCommandKind.Unspecified, reply.Kind);
Assert.NotEqual(ProtocolStatusCode.Unspecified, reply.ProtocolStatus.Code);
Assert.True(reply.HasHresult, $"Fixture '{GetFixtureId(fixture)}' must carry an HRESULT.");
// The malformed-reply and credential-redaction fixtures
// (command-reply.authenticate-user.*) deliberately omit hresult, statuses,
// and/or return_value to exercise the "absent detail" contract paths — see
// docs/ClientBehaviorFixtures.md. The strict MXAccess-reply-detail
// invariants below apply only to fixtures that carry that detail.
if (fixtureId.StartsWith("command-reply.authenticate-user.", StringComparison.Ordinal))
{
continue;
}
Assert.True(reply.HasHresult, $"Fixture '{fixtureId}' must carry an HRESULT.");
Assert.NotEmpty(reply.Statuses);
Assert.NotEqual(MxDataType.Unspecified, reply.ReturnValue.DataType);
Assert.True(
reply.ReturnValue.KindCase != MxValue.KindOneofCase.None || reply.ReturnValue.IsNull,
$"Fixture '{GetFixtureId(fixture)}' must carry a typed value, raw value, or explicit null.");
$"Fixture '{fixtureId}' must carry a typed value, raw value, or explicit null.");
}
MxCommandReply failedWrite = ParseFixture<MxCommandReply>(
@@ -3,22 +3,24 @@ using Google.Protobuf;
using Google.Protobuf.Reflection;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
namespace ZB.MOM.WW.MxGateway.Tests.Contracts;
public sealed class ClientProtoInputTests
{
/// <summary>
/// Guards the published client descriptor set against silent staleness. Every message
/// and field compiled into the in-process contract (which the build regenerates from the current
/// <c>.proto</c> sources) must appear in the committed protoset. A missing symbol means the
/// descriptor was not regenerated after a proto change; run
/// <c>scripts/publish-client-proto-inputs.ps1</c> and commit the refreshed protoset.
/// Guards the published client descriptor set against silent staleness. Every message,
/// field, enum, enum value, service, and method compiled into the in-process contract
/// (which the build regenerates from the current <c>.proto</c> sources) must appear in the
/// committed protoset. A missing symbol means the descriptor was not regenerated after a
/// proto change; run <c>scripts/publish-client-proto-inputs.ps1</c> and commit the
/// refreshed protoset.
/// The check is semantic (symbol presence) rather than byte-wise, so it is independent of protoc
/// version and does not require protoc on the test runner.
/// </summary>
[Fact]
public void Descriptor_ContainsEveryContractMessageAndField()
public void Descriptor_ContainsEveryContractSymbol()
{
DirectoryInfo repositoryRoot = FindRepositoryRoot();
string descriptorPath = Path.Combine(
@@ -34,20 +36,65 @@ public sealed class ClientProtoInputTests
HashSet<string> publishedMessages = new(StringComparer.Ordinal);
HashSet<string> publishedFields = new(StringComparer.Ordinal);
HashSet<string> publishedEnums = new(StringComparer.Ordinal);
HashSet<string> publishedServices = new(StringComparer.Ordinal);
foreach (FileDescriptorProto file in descriptorSet.File)
{
foreach (DescriptorProto message in file.MessageType)
{
CollectPublishedSymbols(file.Package, message, publishedMessages, publishedFields);
CollectPublishedSymbols(file.Package, message, publishedMessages, publishedFields, publishedEnums);
}
foreach (EnumDescriptorProto enumType in file.EnumType)
{
CollectPublishedEnumSymbols(file.Package, enumType, publishedEnums);
}
foreach (ServiceDescriptorProto service in file.Service)
{
string serviceFullName = string.IsNullOrEmpty(file.Package) ? service.Name : file.Package + "." + service.Name;
publishedServices.Add(serviceFullName);
foreach (MethodDescriptorProto method in service.Method)
{
publishedServices.Add(serviceFullName + "/" + method.Name);
}
}
}
List<string> missing = [];
foreach (FileDescriptor file in new[] { MxaccessGatewayReflection.Descriptor, MxaccessWorkerReflection.Descriptor })
FileDescriptor[] contractFiles =
[
MxaccessGatewayReflection.Descriptor,
MxaccessWorkerReflection.Descriptor,
GalaxyRepositoryReflection.Descriptor,
];
foreach (FileDescriptor file in contractFiles)
{
foreach (MessageDescriptor message in file.MessageTypes)
{
CollectMissingContractSymbols(message, publishedMessages, publishedFields, missing);
CollectMissingContractSymbols(message, publishedMessages, publishedFields, publishedEnums, missing);
}
foreach (EnumDescriptor enumType in file.EnumTypes)
{
CollectMissingEnumSymbols(enumType, publishedEnums, missing);
}
foreach (ServiceDescriptor service in file.Services)
{
if (!publishedServices.Contains(service.FullName))
{
missing.Add(service.FullName);
}
foreach (MethodDescriptor method in service.Methods)
{
string key = service.FullName + "/" + method.Name;
if (!publishedServices.Contains(key))
{
missing.Add(key);
}
}
}
}
@@ -62,7 +109,8 @@ public sealed class ClientProtoInputTests
string package,
DescriptorProto message,
HashSet<string> messages,
HashSet<string> fields)
HashSet<string> fields,
HashSet<string> enums)
{
string fullName = string.IsNullOrEmpty(package) ? message.Name : package + "." + message.Name;
messages.Add(fullName);
@@ -74,7 +122,26 @@ public sealed class ClientProtoInputTests
foreach (DescriptorProto nested in message.NestedType)
{
CollectPublishedSymbols(fullName, nested, messages, fields);
CollectPublishedSymbols(fullName, nested, messages, fields, enums);
}
foreach (EnumDescriptorProto enumType in message.EnumType)
{
CollectPublishedEnumSymbols(fullName, enumType, enums);
}
}
private static void CollectPublishedEnumSymbols(
string containingScope,
EnumDescriptorProto enumType,
HashSet<string> enums)
{
string enumFullName = string.IsNullOrEmpty(containingScope) ? enumType.Name : containingScope + "." + enumType.Name;
enums.Add(enumFullName);
foreach (EnumValueDescriptorProto value in enumType.Value)
{
enums.Add(enumFullName + "/" + value.Name);
}
}
@@ -82,6 +149,7 @@ public sealed class ClientProtoInputTests
MessageDescriptor message,
HashSet<string> publishedMessages,
HashSet<string> publishedFields,
HashSet<string> publishedEnums,
List<string> missing)
{
if (!publishedMessages.Contains(message.FullName))
@@ -100,7 +168,32 @@ public sealed class ClientProtoInputTests
foreach (MessageDescriptor nested in message.NestedTypes)
{
CollectMissingContractSymbols(nested, publishedMessages, publishedFields, missing);
CollectMissingContractSymbols(nested, publishedMessages, publishedFields, publishedEnums, missing);
}
foreach (EnumDescriptor enumType in message.EnumTypes)
{
CollectMissingEnumSymbols(enumType, publishedEnums, missing);
}
}
private static void CollectMissingEnumSymbols(
EnumDescriptor enumType,
HashSet<string> publishedEnums,
List<string> missing)
{
if (!publishedEnums.Contains(enumType.FullName))
{
missing.Add(enumType.FullName);
}
foreach (EnumValueDescriptor value in enumType.Values)
{
string key = enumType.FullName + "/" + value.Name;
if (!publishedEnums.Contains(key))
{
missing.Add(key);
}
}
}
@@ -481,6 +481,192 @@ public sealed class WorkerFrameProtocolTests
Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode);
}
/// <summary>
/// Verifies a negotiated frame maximum below the worker floor is rejected as
/// <c>InvalidConfiguration</c> (WRK-24), and that exactly the floor is adopted. The floor matches
/// the gateway's own <c>GatewayOptionsValidator.MinimumMaxMessageBytes</c>, so the worker never
/// rejects a value the gateway's validator accepts, yet a nonsensical tiny value faults at the
/// handshake instead of leaving a session that fails every later frame.
/// </summary>
[Fact]
public void AdoptNegotiatedMaxMessageBytes_BelowFloor_ThrowsInvalidConfiguration()
{
WorkerFrameProtocolOptions belowFloor = CreateOptions();
WorkerFrameProtocolException exception = Assert.Throws<WorkerFrameProtocolException>(
() => belowFloor.AdoptNegotiatedMaxMessageBytes(512));
Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode);
// Boundary: exactly the floor is accepted.
WorkerFrameProtocolOptions atFloor = CreateOptions();
atFloor.AdoptNegotiatedMaxMessageBytes((uint)WorkerFrameProtocolOptions.MinNegotiableFrameBytes);
Assert.Equal(WorkerFrameProtocolOptions.MinNegotiableFrameBytes, atFloor.MaxMessageBytes);
}
/// <summary>
/// WRK-22 / IPC-26. A <c>WriteAsync</c> cancelled while it waits for the write lock must never
/// have its frame written by the next lock-holder. Writer A holds the lock mid-write (blocked in
/// the stream); an event write is queued and then cancelled; when A is released and a later
/// control frame drains, the wire carries A's frame and the control frame only — the cancelled
/// event envelope is tombstoned and skipped.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WriteAsync_CancelledWhileWaitingForLock_FrameIsNeverWritten()
{
WorkerFrameProtocolOptions options = CreateOptions();
using GatedWriteStream stream = new();
WorkerFrameWriter writer = new(stream, options);
// Writer A occupies the writer and blocks inside the stream, holding the write lock.
Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
// Queue an event write with its own CTS while the lock is held, then cancel it.
using CancellationTokenSource cts = new();
Task cancelledWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event, cts.Token);
await Task.Delay(50);
cts.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await cancelledWrite);
// Release A, then drive a fresh control write.
stream.ReleaseFirstWrite();
await AwaitWithTimeoutAsync(firstWrite);
await writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
stream.Position = 0;
WorkerFrameReader reader = new(stream, options);
WorkerEnvelope frame1 = await reader.ReadAsync();
WorkerEnvelope frame2 = await reader.ReadAsync();
Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase);
Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame2.BodyCase);
// The cancelled event never reached the wire — no third frame, and sequences stay contiguous.
Assert.Equal(stream.Length, stream.Position);
Assert.Equal(1UL, frame1.Sequence);
Assert.Equal(2UL, frame2.Sequence);
}
/// <summary>
/// WRK-22 / IPC-26, the review's shutdown scenario. A cancelled event frame queued before a
/// shutdown-ack control frame must not trail the ack on the wire: the tombstone rule plus the
/// control-before-event scheduler keeps the ack the last frame written.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WriteAsync_CancelledEventFrame_DoesNotTrailShutdownAck()
{
WorkerFrameProtocolOptions options = CreateOptions();
using GatedWriteStream stream = new();
WorkerFrameWriter writer = new(stream, options);
Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
using CancellationTokenSource cts = new();
Task cancelledEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event, cts.Token);
await Task.Delay(50);
cts.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await cancelledEvent);
// The shutdown ack (a control frame) is queued behind the still-blocked first write.
Task ackWrite = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
await Task.Delay(50);
stream.ReleaseFirstWrite();
await AwaitWithTimeoutAsync(Task.WhenAll(firstWrite, ackWrite));
stream.Position = 0;
WorkerFrameReader reader = new(stream, options);
WorkerEnvelope frame1 = await reader.ReadAsync();
WorkerEnvelope frame2 = await reader.ReadAsync();
Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase);
// The ack is the last frame — the cancelled event did not trail it.
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase);
Assert.Equal(stream.Length, stream.Position);
}
/// <summary>
/// WRK-25. The batch entry point enqueues a whole event burst under one lock acquisition and
/// drains it together, so N events cost exactly one flush and reach the wire in batch order with
/// monotonic sequences — the coalescing WRK-12 shipped, now on the event hot path.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WriteBatchAsync_FlushesOnceAndPreservesOrder()
{
const int count = 8;
WorkerFrameProtocolOptions options = CreateOptions();
using FlushCountingStream stream = new();
WorkerFrameWriter writer = new(stream, options);
WorkerEnvelope[] batch = new WorkerEnvelope[count];
for (int index = 0; index < count; index++)
{
batch[index] = CreateEventEnvelope(workerSequence: (ulong)(100 + index));
}
await writer.WriteBatchAsync(batch, WorkerFrameWritePriority.Event);
// The whole batch was queued before the single lock wait, so it drained in one pass => one flush.
Assert.Equal(1, stream.FlushCount);
stream.Position = 0;
WorkerFrameReader reader = new(stream, options);
for (int index = 0; index < count; index++)
{
WorkerEnvelope frame = await reader.ReadAsync();
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame.BodyCase);
// Wire order matches batch order.
Assert.Equal((ulong)(100 + index), frame.WorkerEvent.Event.WorkerSequence);
// Write-time stamped sequence is monotonic 1..count.
Assert.Equal((ulong)(index + 1), frame.Sequence);
}
}
/// <summary>
/// WRK-25. Control-before-event still holds mid-batch: a control frame queued while a batch is
/// draining jumps ahead of the batch's remaining events.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WriteBatchAsync_ControlFrameQueuedDuringBatch_JumpsRemainingEvents()
{
WorkerFrameProtocolOptions options = CreateOptions();
using GatedWriteStream stream = new();
WorkerFrameWriter writer = new(stream, options);
WorkerEnvelope[] batch = new[]
{
CreateEventEnvelope(),
CreateEventEnvelope(),
CreateEventEnvelope(),
};
// The batch takes the lock and blocks writing its first event frame inside the stream.
Task batchWrite = writer.WriteBatchAsync(batch, WorkerFrameWritePriority.Event);
await AwaitWithTimeoutAsync(stream.FirstWriteStarted);
// A control frame queued mid-drain must jump the batch's remaining events.
Task controlWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
await Task.Delay(50);
stream.ReleaseFirstWrite();
await AwaitWithTimeoutAsync(Task.WhenAll(batchWrite, controlWrite));
stream.Position = 0;
WorkerFrameReader reader = new(stream, options);
WorkerEnvelope f1 = await reader.ReadAsync();
WorkerEnvelope f2 = await reader.ReadAsync();
WorkerEnvelope f3 = await reader.ReadAsync();
WorkerEnvelope f4 = await reader.ReadAsync();
// First event was already writing when the control frame queued; the control frame then jumps
// ahead of the two remaining events.
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f1.BodyCase);
Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, f2.BodyCase);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f3.BodyCase);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f4.BodyCase);
}
private static WorkerEnvelope CreateGatewayHelloEnvelope(ulong sequence = 1)
{
return new WorkerEnvelope
@@ -522,6 +708,47 @@ public sealed class WorkerFrameProtocolTests
};
}
private static WorkerEnvelope CreateEventEnvelope(ulong workerSequence)
{
WorkerEnvelope envelope = CreateEventEnvelope();
envelope.WorkerEvent.Event.WorkerSequence = workerSequence;
return envelope;
}
private static WorkerEnvelope CreateShutdownAckEnvelope()
{
return new WorkerEnvelope
{
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
SessionId = SessionId,
WorkerShutdownAck = new WorkerShutdownAck
{
Status = new ProtocolStatus
{
Code = ProtocolStatusCode.Ok,
Message = "OK",
},
},
};
}
// A MemoryStream that counts FlushAsync calls without gating any write, so a batch write can be
// asserted to flush exactly once.
private sealed class FlushCountingStream : MemoryStream
{
private int _flushCount;
/// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
public int FlushCount => Volatile.Read(ref _flushCount);
/// <inheritdoc />
public override Task FlushAsync(CancellationToken cancellationToken)
{
Interlocked.Increment(ref _flushCount);
return base.FlushAsync(cancellationToken);
}
}
// A MemoryStream whose first WriteAsync blocks until released, so a test can queue additional frames
// behind an in-progress write and observe the writer's priority ordering.
private sealed class GatedWriteStream : MemoryStream
@@ -1112,6 +1112,184 @@ public sealed class WorkerPipeSessionTests
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
/// <summary>
/// WRK-27. An STA call outside the command dispatcher (the alarm poll) advertises itself on
/// the heartbeat snapshot's <c>StaCallInProgress</c> flag, and the watchdog grants it the same
/// grace-to-ceiling suppression as a dispatched command: stale STA activity within the ceiling
/// does not fault while the flag is set, but stale activity beyond the ceiling faults anyway.
/// This closes the 15 s-vs-75 s asymmetry between polls and commands.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Watchdog_StaCallInProgress_SuppressedUntilCeiling()
{
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10));
// Phase 1 — within the ceiling: stale beyond grace, empty correlation id, StaCallInProgress
// set. The default 75 s ceiling is far beyond the 5 s staleness, so the watchdog must suppress.
using (PipePair pipePair = await PipePair.CreateAsync(cancellation.Token))
{
FakeRuntimeSession runtime = new();
runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow - TimeSpan.FromSeconds(5),
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId: string.Empty,
staCallInProgress: true));
WorkerPipeSession session = CreatePipeSession(
pipePair.WorkerStream,
runtime,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
HeartbeatGrace = TimeSpan.FromMilliseconds(50),
});
Task runTask = session.RunAsync(cancellation.Token);
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
const int framesToInspect = 6;
int heartbeatsObserved = 0;
for (int index = 0; index < framesToInspect; index++)
{
WorkerEnvelope envelope = await pipePair.GatewayReader.ReadAsync(cancellation.Token);
Assert.NotEqual(WorkerEnvelope.BodyOneofCase.WorkerFault, envelope.BodyCase);
if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerHeartbeat)
{
heartbeatsObserved++;
}
}
Assert.True(
heartbeatsObserved >= 2,
$"Expected multiple heartbeats during the in-progress STA-call window; observed {heartbeatsObserved}.");
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
// Phase 2 — beyond the ceiling: same StaCallInProgress flag, but staleness (5 s) exceeds the
// 200 ms ceiling, so the watchdog must fire even with the poll in progress.
using (PipePair pipePair = await PipePair.CreateAsync(cancellation.Token))
{
FakeRuntimeSession runtime = new();
runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset.UtcNow - TimeSpan.FromSeconds(5),
pendingCommandCount: 0,
outboundEventQueueDepth: 0,
lastEventSequence: 0,
currentCommandCorrelationId: string.Empty,
staCallInProgress: true));
WorkerPipeSession session = CreatePipeSession(
pipePair.WorkerStream,
runtime,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
HeartbeatGrace = TimeSpan.FromMilliseconds(50),
HeartbeatStuckCeiling = TimeSpan.FromMilliseconds(200),
});
Task runTask = session.RunAsync(cancellation.Token);
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
WorkerEnvelope fault = await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerFault,
cancellation.Token);
Assert.Equal(WorkerFaultCategory.StaHung, fault.WorkerFault.Category);
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
}
/// <summary>
/// WRK-25. The event drain loop submits a whole drained batch through the writer's batch entry
/// point, so a burst of 128 events costs one flush, not 128 — the assertion the WRK-12
/// tracking claim needed to actually hold on the event hot path.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task EventBurst_DrainLoopCoalescesFlushes()
{
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15));
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
FakeRuntimeSession runtime = new();
// A far-off heartbeat interval keeps heartbeat flushes out of the measurement window.
FlushCountingPassthroughStream countingStream = new(pipePair.WorkerStream);
WorkerPipeSession session = CreatePipeSession(
countingStream,
runtime,
new WorkerPipeSessionOptions
{
HeartbeatInterval = TimeSpan.FromMinutes(5),
HeartbeatGrace = TimeSpan.FromSeconds(30),
});
Task runTask = session.RunAsync(cancellation.Token);
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
// Let the idle drain loop settle (no events yet → no flushes) and record the baseline.
await Task.Delay(100, cancellation.Token);
int baselineFlushes = countingStream.FlushCount;
// Enqueue a full 128-event batch atomically so the drain loop sees it as one batch.
const int burst = 128;
List<WorkerEvent> batch = new(burst);
for (int index = 0; index < burst; index++)
{
batch.Add(CreateWorkerEvent(sequence: (ulong)(index + 1)));
}
runtime.EnqueueEvents(batch);
// Drain all 128 events off the gateway side.
for (int index = 0; index < burst; index++)
{
await ReadUntilAsync(
pipePair.GatewayReader,
WorkerEnvelope.BodyOneofCase.WorkerEvent,
cancellation.Token);
}
// The whole burst cost exactly one additional flush.
Assert.Equal(1, countingStream.FlushCount - baselineFlushes);
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
}
/// <summary>
/// WRK-24. A GatewayHello negotiating a frame maximum below the worker floor faults at the
/// handshake with a fault frame rather than being adopted — mirroring the above-ceiling
/// handshake behavior — so a nonsensical tiny value never leaves a session that fails every
/// later frame. No message loop is entered.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Handshake_GatewayHelloWithTinyMaxFrameBytes_FaultsAtHandshake()
{
WorkerFrameProtocolOptions options = CreateOptions();
using MemoryStream inbound = new();
await new WorkerFrameWriter(inbound, options)
.WriteAsync(CreateGatewayHelloEnvelope(maxFrameBytes: 512));
inbound.Position = 0;
using MemoryStream outbound = new();
WorkerPipeSession session = CreateSession(inbound, outbound, options);
bool initialized = false;
WorkerFrameProtocolException exception =
await Assert.ThrowsAsync<WorkerFrameProtocolException>(
async () => await session.CompleteStartupHandshakeAsync(
_ =>
{
initialized = true;
return Task.CompletedTask;
}));
Assert.False(initialized);
Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode);
WorkerEnvelope fault = Assert.Single(ReadWrittenFrames(outbound, options));
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerFault, fault.BodyCase);
}
/// <summary>
/// Regression test: a long in-flight STA command that keeps pumping
/// must NOT self-fault as <c>StaHung</c>, and its reply must still be
@@ -1976,6 +2154,78 @@ public sealed class WorkerPipeSessionTests
}
}
// Wraps the worker side of the pipe and counts FlushAsync calls so a test can assert the event
// drain loop coalesces a burst into a single flush. Delegates every other operation to the inner
// stream; does not own the inner stream's lifetime (PipePair disposes it).
private sealed class FlushCountingPassthroughStream : Stream
{
private readonly Stream inner;
private int flushCount;
/// <summary>Initializes the passthrough over the given inner stream.</summary>
/// <param name="inner">The stream to delegate to.</param>
public FlushCountingPassthroughStream(Stream inner)
{
this.inner = inner;
}
/// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
public int FlushCount => Volatile.Read(ref flushCount);
/// <inheritdoc />
public override bool CanRead => inner.CanRead;
/// <inheritdoc />
public override bool CanSeek => inner.CanSeek;
/// <inheritdoc />
public override bool CanWrite => inner.CanWrite;
/// <inheritdoc />
public override long Length => inner.Length;
/// <inheritdoc />
public override long Position
{
get => inner.Position;
set => inner.Position = value;
}
/// <inheritdoc />
public override void Flush()
{
Interlocked.Increment(ref flushCount);
inner.Flush();
}
/// <inheritdoc />
public override Task FlushAsync(CancellationToken cancellationToken)
{
Interlocked.Increment(ref flushCount);
return inner.FlushAsync(cancellationToken);
}
/// <inheritdoc />
public override int Read(byte[] buffer, int offset, int count) => inner.Read(buffer, offset, count);
/// <inheritdoc />
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
=> inner.ReadAsync(buffer, offset, count, cancellationToken);
/// <inheritdoc />
public override long Seek(long offset, SeekOrigin origin) => inner.Seek(offset, origin);
/// <inheritdoc />
public override void SetLength(long value) => inner.SetLength(value);
/// <inheritdoc />
public override void Write(byte[] buffer, int offset, int count) => inner.Write(buffer, offset, count);
/// <inheritdoc />
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
=> inner.WriteAsync(buffer, offset, count, cancellationToken);
}
private sealed class PipePair : IDisposable
{
private readonly NamedPipeServerStream gatewayStream;
@@ -438,6 +438,53 @@ public sealed class MxAccessStaSessionTests
Assert.Contains("alarm poll failed", fault.DiagnosticMessage, StringComparison.OrdinalIgnoreCase);
}
/// <summary>
/// WRK-27. While the alarm poll's PollOnce is executing on the STA, a heartbeat captured mid-poll
/// must report <see cref="WorkerRuntimeHeartbeatSnapshot.StaCallInProgress"/> so the watchdog
/// grants the poll the same grace-to-ceiling suppression as a dispatched command; once the poll
/// returns the flag clears. PollOnce is blocked on a gate so the heartbeat can be captured while
/// the STA call is genuinely in flight.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task CaptureHeartbeat_DuringAlarmPoll_ReportsStaCallInProgress()
{
FakeAlarmCommandHandler handler = new() { BlockPoll = true };
FakeMxAccessComObjectFactory factory = new();
FakeMxAccessEventSink eventSink = new();
using StaRuntime runtime = CreateRuntime();
using MxAccessStaSession session = new(
runtime,
factory,
eventSink,
new MxAccessEventQueue(),
(_eq, _affinity, _comFactory) => handler);
await session.StartAsync("session-1", workerProcessId: 1);
// Wait until PollOnce is blocked mid-call on the STA thread.
Assert.True(
handler.WaitForPollEntered(TimeSpan.FromSeconds(5)),
"Expected the alarm poll to start within 5 seconds.");
// Captured mid-poll, the heartbeat advertises the in-progress STA call.
Assert.True(session.CaptureHeartbeat().StaCallInProgress);
// Release the poll and stop blocking; the flag clears once the poll returns.
handler.BlockPoll = false;
handler.ReleasePoll();
using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(5));
while (session.CaptureHeartbeat().StaCallInProgress && !timeout.IsCancellationRequested)
{
await Task.Delay(25, CancellationToken.None);
}
Assert.False(
session.CaptureHeartbeat().StaCallInProgress,
"Expected StaCallInProgress to clear once the alarm poll returned.");
}
/// <summary>
/// The STA-affinity guard throws when an
/// IMxAccessAlarmConsumer call is attempted off the thread that created
@@ -472,6 +519,8 @@ public sealed class MxAccessStaSessionTests
private sealed class FakeAlarmCommandHandler : IAlarmCommandHandler
{
private readonly object gate = new object();
private readonly ManualResetEventSlim pollEntered = new(false);
private readonly ManualResetEventSlim releasePoll = new(false);
private int pollCount;
private int? lastPollThreadId;
@@ -484,6 +533,17 @@ public sealed class MxAccessStaSessionTests
/// <summary>Exception thrown by PollOnce; null to succeed.</summary>
public Exception? PollException { get; set; }
/// <summary>When set, <see cref="PollOnce"/> blocks until <see cref="ReleasePoll"/> is called.</summary>
public bool BlockPoll { get; set; }
/// <summary>Waits until a blocking <see cref="PollOnce"/> has entered and is blocked.</summary>
/// <param name="timeout">Maximum time to wait.</param>
/// <returns>True if a poll entered within the timeout.</returns>
public bool WaitForPollEntered(TimeSpan timeout) => pollEntered.Wait(timeout);
/// <summary>Releases a <see cref="PollOnce"/> blocked on the gate.</summary>
public void ReleasePoll() => releasePoll.Set();
/// <summary>Gets the count of PollOnce calls.</summary>
public int PollCount
{
@@ -533,6 +593,12 @@ public sealed class MxAccessStaSessionTests
lastPollThreadId = Thread.CurrentThread.ManagedThreadId;
}
if (BlockPoll)
{
pollEntered.Set();
releasePoll.Wait(TimeSpan.FromSeconds(10));
}
if (PollException is not null)
{
throw PollException;
@@ -540,6 +606,10 @@ public sealed class MxAccessStaSessionTests
}
/// <inheritdoc />
public void Dispose() { }
public void Dispose()
{
pollEntered.Dispose();
releasePoll.Dispose();
}
}
}
@@ -372,6 +372,23 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
}
}
/// <summary>
/// Enqueues a batch of worker events atomically under one lock so the drain loop cannot
/// observe a partial batch. Lets a test assert the drain loop coalesces a whole batch into one
/// flush (WRK-25) without racing a mid-enqueue drain that would split the batch.
/// </summary>
/// <param name="workerEvents">The events to enqueue in order.</param>
public void EnqueueEvents(IEnumerable<WorkerEvent> workerEvents)
{
lock (gate)
{
foreach (WorkerEvent workerEvent in workerEvents)
{
events.Enqueue(workerEvent);
}
}
}
/// <inheritdoc />
public void Dispose()
{
@@ -18,6 +18,17 @@ public sealed class WorkerFrameProtocolOptions
/// </summary>
public const int MaxNegotiableFrameBytes = 256 * 1024 * 1024;
/// <summary>
/// Lower floor the worker will accept for a gateway-negotiated frame maximum
/// (<c>GatewayHello.max_frame_bytes</c>). Matches the gateway's own
/// <c>GatewayOptionsValidator.MinimumMaxMessageBytes</c> validation floor so the worker never
/// rejects a value the gateway's own validator accepts as legal configuration, yet a nonsensical
/// tiny value (a gateway bug or a foreign/old peer) is rejected at the handshake rather than
/// leaving a session that handshakes cleanly and then fails every subsequent frame with
/// per-frame size errors. 1024 bytes still guarantees hellos, heartbeats, acks, and faults fit.
/// </summary>
public const int MinNegotiableFrameBytes = 1024;
/// <summary>Initializes a new instance of the WorkerFrameProtocolOptions class from WorkerOptions.</summary>
/// <param name="options">Worker initialization options.</param>
public WorkerFrameProtocolOptions(WorkerOptions options)
@@ -118,7 +129,9 @@ public sealed class WorkerFrameProtocolOptions
/// <summary>
/// Adopts the gateway-negotiated frame maximum conveyed in <c>GatewayHello.max_frame_bytes</c>.
/// A value of 0 (an older gateway that never set the field) is ignored and the
/// constructor default is kept. A value above <see cref="MaxNegotiableFrameBytes"/> is rejected.
/// constructor default is kept. A value outside the accepted range
/// [<see cref="MinNegotiableFrameBytes"/>, <see cref="MaxNegotiableFrameBytes"/>] is rejected so a
/// nonsensical negotiated value faults at the handshake rather than mid-session.
/// </summary>
/// <param name="negotiatedMaxFrameBytes">The gateway-negotiated maximum, or 0 for "keep default".</param>
internal void AdoptNegotiatedMaxMessageBytes(uint negotiatedMaxFrameBytes)
@@ -128,6 +141,14 @@ public sealed class WorkerFrameProtocolOptions
return;
}
if (negotiatedMaxFrameBytes < MinNegotiableFrameBytes)
{
throw new WorkerFrameProtocolException(
WorkerFrameProtocolErrorCode.InvalidConfiguration,
$"GatewayHello negotiated frame maximum {negotiatedMaxFrameBytes} is below the worker floor "
+ $"of {MinNegotiableFrameBytes} bytes.");
}
if (negotiatedMaxFrameBytes > MaxNegotiableFrameBytes)
{
throw new WorkerFrameProtocolException(
@@ -1,6 +1,7 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
using Google.Protobuf;
@@ -33,6 +34,16 @@ public sealed class WorkerFrameWriter
/// <summary>Gets the completion source signaled once the frame has been written or has failed.</summary>
public TaskCompletionSource<bool> Completion { get; }
/// <summary>
/// Set to <c>true</c> by <see cref="DequeueNext"/> — under <c>_gate</c> — at the instant the
/// draining lock-holder takes ownership of this frame to write it. A cancelled caller
/// tombstones its frame only while it is still unclaimed, so a claim and a cancel can never
/// both win: the flag is the interlock between the two. A frame already claimed is mid-write
/// and can no longer be recalled (see <see cref="WriteAsync(WorkerEnvelope, WorkerFrameWritePriority, CancellationToken)"/>).
/// Mutated only under <c>_gate</c>.
/// </summary>
public bool Claimed;
}
private readonly WorkerFrameProtocolOptions _options;
@@ -75,6 +86,13 @@ public sealed class WorkerFrameWriter
/// <param name="priority">Scheduling priority; control frames are written ahead of event frames.</param>
/// <param name="cancellationToken">Token to cancel waiting for the write lock.</param>
/// <returns>A task that completes when the frame has been written and flushed.</returns>
/// <remarks>
/// Cancellation contract (WRK-22): if the token fires while this call is waiting for the write
/// lock, the frame is tombstoned so it is never written — unless a draining lock-holder has
/// already claimed it, in which case the frame may still reach the wire even though this call
/// observes <see cref="OperationCanceledException"/>. That residual window is by design: blocking
/// the canceller behind the very write it is abandoning would defeat the point of cancellation.
/// </remarks>
public async Task WriteAsync(
WorkerEnvelope envelope,
WorkerFrameWritePriority priority,
@@ -101,7 +119,19 @@ public sealed class WorkerFrameWriter
// Contend for the single writer: whoever wins drains every currently-queued frame in priority
// order, so this frame is written by this call or by a concurrent caller that got the lock
// first. Either way it completes via its own TaskCompletionSource.
await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
// Tombstone the queued frame so DequeueNext skips it — but only if a draining lock-holder
// has not already claimed it. If it is claimed it is mid-write and cannot be recalled; the
// caller still observes cancellation while the frame reaches the wire (documented above).
TombstoneIfUnclaimed(frame, cancellationToken);
throw;
}
try
{
await DrainQueuedFramesAsync().ConfigureAwait(false);
@@ -114,6 +144,123 @@ public sealed class WorkerFrameWriter
await frame.Completion.Task.ConfigureAwait(false);
}
/// <summary>
/// Queues a whole batch of envelopes at one priority under a single lock acquisition and drains
/// it, so a burst of frames — the event drain loop's hot path — pays one flush for the batch
/// rather than one per frame (WRK-25, realizing the WRK-12 coalescing on the path it was built
/// for). Intra-batch order is preserved because the enqueue is atomic under <c>_gate</c> and each
/// class queue is FIFO; the control-before-event guarantee still holds because any concurrently
/// queued control frame is drained ahead of this batch by <see cref="DequeueNext"/>. Every frame's
/// "written and flushed before completion" contract is unchanged.
/// </summary>
/// <param name="envelopes">Envelopes to write, in order.</param>
/// <param name="priority">Scheduling priority for the whole batch.</param>
/// <param name="cancellationToken">Token to cancel waiting for the write lock.</param>
/// <returns>A task that completes when every frame in the batch has been written and flushed.</returns>
/// <remarks>
/// A per-frame rejection inside the batch (for example one oversized event) surfaces from the
/// awaited completions as its <see cref="WorkerFrameProtocolException"/>; the remaining frames are
/// still observed so none faults unobserved. Cancellation while waiting for the lock tombstones
/// every still-unclaimed frame in the batch, per the WRK-22 contract on
/// <see cref="WriteAsync(WorkerEnvelope, WorkerFrameWritePriority, CancellationToken)"/>.
/// </remarks>
public async Task WriteBatchAsync(
IReadOnlyList<WorkerEnvelope> envelopes,
WorkerFrameWritePriority priority,
CancellationToken cancellationToken = default)
{
if (envelopes is null)
{
throw new ArgumentNullException(nameof(envelopes));
}
if (envelopes.Count == 0)
{
return;
}
PendingFrame[] frames = new PendingFrame[envelopes.Count];
for (int index = 0; index < envelopes.Count; index++)
{
WorkerEnvelope envelope = envelopes[index]
?? throw new ArgumentException("Batch envelopes must not contain null.", nameof(envelopes));
frames[index] = new PendingFrame(envelope);
}
lock (_gate)
{
Queue<PendingFrame> queue = priority == WorkerFrameWritePriority.Event ? _eventFrames : _controlFrames;
foreach (PendingFrame frame in frames)
{
queue.Enqueue(frame);
}
}
try
{
await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
TombstoneUnclaimed(frames, cancellationToken);
throw;
}
try
{
await DrainQueuedFramesAsync().ConfigureAwait(false);
}
finally
{
_writeLock.Release();
}
// Await every completion so no per-frame rejection faults unobserved, but surface the first
// failure (in batch order) to the caller — the drain loop maps it back to the offending event.
Exception? firstFailure = null;
foreach (PendingFrame frame in frames)
{
try
{
await frame.Completion.Task.ConfigureAwait(false);
}
catch (Exception exception)
{
firstFailure ??= exception;
}
}
if (firstFailure is not null)
{
ExceptionDispatchInfo.Capture(firstFailure).Throw();
}
}
private void TombstoneIfUnclaimed(PendingFrame frame, CancellationToken cancellationToken)
{
lock (_gate)
{
if (!frame.Claimed)
{
frame.Completion.TrySetCanceled(cancellationToken);
}
}
}
private void TombstoneUnclaimed(PendingFrame[] frames, CancellationToken cancellationToken)
{
lock (_gate)
{
foreach (PendingFrame frame in frames)
{
if (!frame.Claimed)
{
frame.Completion.TrySetCanceled(cancellationToken);
}
}
}
}
// Runs only under _writeLock. Drains control frames before event frames, stamping and writing each.
// The stream write itself is not cancellable: a frame is written atomically or fails, never left
// half-written on the pipe because a caller gave up waiting.
@@ -198,18 +345,36 @@ public sealed class WorkerFrameWriter
or WorkerFrameProtocolErrorCode.SessionMismatch;
}
// Returns the next frame to write, control frames first, skipping any frame a cancelled caller
// tombstoned while it waited for the lock (WRK-22). The frame actually returned is marked Claimed
// under _gate in the same critical section that checks the tombstone, so a claim and a concurrent
// cancel are mutually exclusive: whichever acquires _gate first wins.
private PendingFrame? DequeueNext()
{
lock (_gate)
{
if (_controlFrames.Count > 0)
while (_controlFrames.Count > 0)
{
return _controlFrames.Dequeue();
PendingFrame frame = _controlFrames.Dequeue();
if (frame.Completion.Task.IsCanceled)
{
continue;
}
frame.Claimed = true;
return frame;
}
if (_eventFrames.Count > 0)
while (_eventFrames.Count > 0)
{
return _eventFrames.Dequeue();
PendingFrame frame = _eventFrames.Dequeue();
if (frame.Completion.Task.IsCanceled)
{
continue;
}
frame.Claimed = true;
return frame;
}
return null;
@@ -371,27 +371,57 @@ public sealed class WorkerPipeSession
continue;
}
foreach (WorkerEvent workerEvent in events)
// Submit the whole drained batch through the writer's batch entry point under one lock
// acquisition so the burst pays a single flush instead of one per event (WRK-25). Events
// are the low-priority frame class: the writer holds them behind any pending control frame
// (reply, fault, heartbeat, shutdown ack) so those are not delayed behind an event backlog,
// and intra-batch order is preserved.
WorkerEnvelope[] envelopes = new WorkerEnvelope[events.Count];
for (int index = 0; index < events.Count; index++)
{
// Events are the low-priority frame class: the writer holds them behind any pending
// control frame (reply, fault, heartbeat, shutdown ack) so those are not delayed
// behind an event backlog.
try
{
await _writer
.WriteAsync(CreateEnvelope(workerEvent), WorkerFrameWritePriority.Event, cancellationToken)
.ConfigureAwait(false);
}
catch (WorkerFrameProtocolException exception)
when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge)
{
await FaultOnOversizedEventAsync(workerEvent, exception, cancellationToken)
.ConfigureAwait(false);
}
envelopes[index] = CreateEnvelope(events[index]);
}
try
{
await _writer
.WriteBatchAsync(envelopes, WorkerFrameWritePriority.Event, cancellationToken)
.ConfigureAwait(false);
}
catch (WorkerFrameProtocolException exception)
when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge)
{
// A single oversized event surfaces from the batch's awaited completions; the death is
// still IPC-30's structured, event-naming fault. Map the rejection back to the first
// event in batch order whose envelope overshoots the negotiated maximum — the same
// frame the writer rejected first.
await FaultOnOversizedEventAsync(
FindOversizedEvent(events, envelopes),
exception,
cancellationToken)
.ConfigureAwait(false);
}
}
}
private WorkerEvent FindOversizedEvent(
IReadOnlyList<WorkerEvent> events,
WorkerEnvelope[] envelopes)
{
for (int index = 0; index < envelopes.Length; index++)
{
if (envelopes[index].CalculateSize() > _options.MaxMessageBytes)
{
return events[index];
}
}
// Unreachable in practice: WriteBatchAsync surfaced MessageTooLarge, so at least one envelope
// exceeded the negotiated maximum. Fall back to the first event so the fault still names a
// concrete event rather than throwing a second, less useful exception from the fault path.
return events[0];
}
/// <summary>
/// Ends the session on an event that cannot be framed, but deliberately and diagnosably
/// (IPC-30). An event above the negotiated frame maximum is undeliverable end to end — the
@@ -1085,16 +1115,16 @@ public sealed class WorkerPipeSession
return;
}
if (!string.IsNullOrEmpty(snapshot.CurrentCommandCorrelationId)
if ((!string.IsNullOrEmpty(snapshot.CurrentCommandCorrelationId) || snapshot.StaCallInProgress)
&& staleFor <= _sessionOptions.HeartbeatStuckCeiling)
{
// A command is in flight and we are still within the defensive
// suppression ceiling — the STA is busy executing it, not
// hung. The next MarkActivity() in StaRuntime.ProcessQueuedCommands
// will refresh LastActivityUtc once the command returns, at which
// point this branch stops being taken. The heartbeat already
// surfaces the in-flight correlation id so the gateway can apply
// its own per-command timeout if it considers the command too slow.
// A command is in flight, or an STA call outside the dispatcher (the alarm poll, WRK-27) is
// executing, and we are still within the defensive suppression ceiling — the STA is busy
// doing that work, not hung. The next MarkActivity() in StaRuntime.ProcessQueuedCommands
// will refresh LastActivityUtc once the work returns, at which point this branch stops
// being taken. The heartbeat already surfaces the in-flight correlation id so the gateway
// can apply its own per-command timeout if it considers the command too slow; a poll that
// blocks the STA past the ceiling still faults, which is the ceiling's contract.
return;
}
@@ -24,6 +24,14 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
private CancellationTokenSource? alarmPollCts;
private Task? alarmPollTask;
private int? alarmConsumerThreadId;
// True on the STA thread exactly around the alarm PollOnce COM call. The alarm poll runs outside
// the StaCommandDispatcher (so it does not inflate PendingCommandCount or perturb command dispatch
// ordering), which means CaptureHeartbeat would otherwise see no in-flight activity during a long
// poll and the watchdog would fault the session at the 15 s grace instead of the 75 s ceiling
// granted to dispatched commands. Surfacing the poll on the heartbeat closes that asymmetry
// (WRK-27). Volatile: written on the STA thread, read on the heartbeat thread.
private volatile bool staAlarmPollInProgress;
private bool disposed;
/// <summary>
@@ -247,8 +255,20 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
await staRuntime.InvokeAsync(
() =>
{
EnsureOnAlarmConsumerThread();
handler.PollOnce();
// Advertise the poll to the watchdog for exactly the span of the COM call
// (WRK-27): set on the STA thread immediately before the affinity check and
// PollOnce, clear in the finally so a heartbeat captured mid-poll reports
// StaCallInProgress and one captured after does not.
staAlarmPollInProgress = true;
try
{
EnsureOnAlarmConsumerThread();
handler.PollOnce();
}
finally
{
staAlarmPollInProgress = false;
}
},
cancellationToken).ConfigureAwait(false);
}
@@ -377,7 +397,8 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
pendingCommandCount,
(uint)eventQueue.Count,
eventQueue.LastEventSequence,
currentCommandCorrelationId);
currentCommandCorrelationId,
staAlarmPollInProgress);
}
/// <inheritdoc />
@@ -10,18 +10,26 @@ public sealed class WorkerRuntimeHeartbeatSnapshot
/// <param name="outboundEventQueueDepth">Current depth of the worker event queue.</param>
/// <param name="lastEventSequence">Sequence number of the most recent event.</param>
/// <param name="currentCommandCorrelationId">Correlation ID of the in-flight command.</param>
/// <param name="staCallInProgress">
/// True while an STA call outside the command dispatcher is executing (currently the alarm poll,
/// WRK-27). The watchdog treats this like an in-flight command: it suppresses the stale-STA fault
/// up to the stuck ceiling instead of the shorter grace, so a healthy-but-slow poll does not fault
/// a healthy session. Named generically so any future non-dispatcher STA work reuses it.
/// </param>
public WorkerRuntimeHeartbeatSnapshot(
DateTimeOffset lastStaActivityUtc,
uint pendingCommandCount,
uint outboundEventQueueDepth,
ulong lastEventSequence,
string currentCommandCorrelationId)
string currentCommandCorrelationId,
bool staCallInProgress = false)
{
LastStaActivityUtc = lastStaActivityUtc;
PendingCommandCount = pendingCommandCount;
OutboundEventQueueDepth = outboundEventQueueDepth;
LastEventSequence = lastEventSequence;
CurrentCommandCorrelationId = currentCommandCorrelationId ?? string.Empty;
StaCallInProgress = staCallInProgress;
}
/// <summary>Gets the last STA activity timestamp in UTC.</summary>
@@ -38,4 +46,11 @@ public sealed class WorkerRuntimeHeartbeatSnapshot
/// <summary>Gets the correlation ID of the in-flight command.</summary>
public string CurrentCommandCorrelationId { get; }
/// <summary>
/// Gets a value indicating whether an STA call outside the command dispatcher (the alarm poll) is
/// executing. When true the watchdog grants the poll the same grace-to-ceiling suppression as a
/// dispatched command (WRK-27).
/// </summary>
public bool StaCallInProgress { get; }
}