diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 73bdcb8..765f3fc 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -40,9 +40,9 @@ Sequenced by cluster; a cluster is one change set. | ID | Sev | Eff | Dep | Status | Title | |---|---|:-:|---|---|---| -| GWC-25 | Medium | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | -| CLI-35 | Medium | S | GWC-25 (coord) | Not started | Python CLI `stream-events` crashes on a ReplayGap | -| CLI-36 | Medium | S | GWC-25 (coord) | Not started | Go CLI `stream-events` silently destroys the ReplayGap signal | +| GWC-25 | Medium | S | CLI-35/36 (coord) | Done | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | +| CLI-35 | Medium | S | GWC-25 (coord) | Done | Python CLI `stream-events` crashes on a ReplayGap | +| CLI-36 | Medium | S | GWC-25 (coord) | Done | Go CLI `stream-events` silently destroys the ReplayGap signal | | WRK-21 | Medium | M | owns IPC-23 fix; WRK-28 same batch | Not started | DrainEvents bound is count-based only; oversized reply kills the session and loses the drained events | | IPC-23 | Medium | S | WRK-21 | Not started | DrainEvents contract requirements (reply fits negotiated max, no event loss, drain-until-empty) + proto-comment/doc wave | | IPC-30 | Low | M | WRK-21 (same batch) | Not started | Oversized event frame stays session-fatal by design, but the death becomes structured (fault frame + logged identity) | @@ -60,7 +60,7 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| | GWC-24 | Medium | P1 | M | GWC-21 (coord, old tracker) | Not started | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | -| GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0` | +| GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Done | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0` | | GWC-26 | Low | P2 | M | GWC-27 | Not started | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed | | GWC-27 | Low | P2 | S | GWC-26 | Not started | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor | | GWC-28 | Low | P2 | S | GWC-10 (coord, old tracker) | Not started | Gateway→worker envelope `sequence` stamped at creation, not at write | @@ -110,8 +110,8 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| CLI-35 | Medium | P0 | S | GWC-25 (coord) | Not started | Python CLI `stream-events` crashes on a ReplayGap | -| CLI-36 | Medium | P0 | S | GWC-25 (coord) | Not started | Go CLI `stream-events` silently destroys the ReplayGap signal | +| CLI-35 | Medium | P0 | S | GWC-25 (coord) | Done | Python CLI `stream-events` crashes on a ReplayGap | +| CLI-36 | Medium | P0 | S | GWC-25 (coord) | Done | Go CLI `stream-events` silently destroys the ReplayGap signal | | CLI-37 | Medium | P1 | M | CLI-38 (co-land) | Not started | Status-array validation must branch on `category` per the proto contract (4-vs-1 divergence) | | CLI-38 | Medium | P1 | S | — | Not started | Align .NET/Go/Java on `hresult < 0` — lands old CLI-08, cures design-doc drift | | CLI-39 | Medium | P1 | S | CLI-35..38, CLI-45 (land last) | Not started | Bump client versions off published 0.1.2 (converge on 0.2.0); registry-collision guard in pack-clients.ps1 | @@ -161,3 +161,4 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-07-13 | Operator bring-up complete: dedicated CI ed25519 key installed in windev `administrators_authorized_keys` (authorized into `dohertj2`, which owns the working MXAccess/toolchain env — a fresh OS account would break the build; the key is independently revocable), Gitea secrets `WINDEV_SSH_KEY`/`WINDEV_SSH_KNOWN_HOSTS` + variable `WINDEV_SSH_USER=dohertj2` stored, runner→`10.100.0.48:22` egress verified on the `traefik` net, issue-write confirmed. **TST-25/TST-26 → `Done`:** credentialed `windows-x86` ran GREEN on `d769244` (Gitea run #37) — Linux runner SSHed windev, checked out the SHA in `C:\build\mxaccessgw-ci` under lock, ran the x86 Worker build + `Worker.Tests`, exit 0; `nightly-windev` correctly skipped on the push event. Branch merged to `main`. Follow-ups (old tracker): revisit **TST-05** (scheduled live smoke — now covered by `nightly-windev`) and **TST-24** (client wire tests) which this unlocks. | | 2026-07-13 | Ran the TST-25 acceptance checks (scripts/ci/README.md) — they caught **two real CI defects, both fixed** on `fix/tst-25-ci-key-log-leak`: (1) **CI SSH key leaked in cleartext** in the `windows-x86` step env echo (Gitea's line-oriented masker missed the multiline PEM) — rotated the CI key on windev (old pubkey revoked), stored the key **base64-encoded** so the masker redacts it to `***` (confirmed on run #38), taught `run-windev-ci.sh` to decode, dropped the redundant public known-hosts secret from the job env; (2) **bootstrap lock race** — `run-windev-ci.sh`'s pre-hand-off `git fetch`/`checkout` ran outside the worktree lock, so concurrent runs collided on `.git/index.lock`; the bootstrap now holds the lock (ps1 re-uses it via `MXGW_CI_LOCK_HELD`), retest confirmed clean serialization. Also **deflaked** `SessionManagerTests` fail-fast timing assertions (absolute `<100ms` wall-clock bound flaked under CI load; now anchored to the configured timeout / dropped for the zero-timeout case). Checks passed: unreachable-host fast-fail (exit 255/15s), deliberate-red propagation (Worker.Tests failure → exit 1), lock concurrency (2nd run waits), no-key-in-logs (masked). Merge target `df7e20d` verified GREEN via the local windev path (Worker build + 356 tests); merged to `main` `19cbf7b`. Check 6 (forced-failure nightly issue): issue endpoint+token proven live at bring-up (#124); in-CI forced-failure probe abandoned to shared-runner congestion (residual `if: failure()` gating is standard Actions). | | 2026-07-13 | New finding **TST-30** (`Low`/`P2`) added — surfaced during TST-25 acceptance: CI runs on a single shared `gitea-runner` (`maxParallel=1`, co-located `10.100.0.35`) interleaved with `dohertj2/lmxopcua`, and Gitea 1.26 exposes no run cancel/delete, so queue latency is unbounded under cross-repo contention and the runner is a single point of failure. Design: add a second/labelled runner + document the no-cancel reality and the `run-windev-ci.sh` bypass. Roll-ups updated (Testing Low 2→3, total 47→48; P2 9→10). | +| 2026-08-07 | **ReplayGap end-to-end cluster (GWC-25 + CLI-35 + CLI-36) → `Done`** on `fix/gwc-25-replaygap-trio`. GWC-25: `SessionEventDistributor.RegisterWithReplay`'s empty-ring branch now reports `oldestAvailableSequence = _highestSequenceSeen + 1` when `gap == true` (still `0` when no gap), so the universal `oldest - 1` resume formula no longer wraps to `ulong.MaxValue` and dead-stream the subscriber; `docs/Sessions.md` documents the empty-ring value. CLI-35: the Python CLI renders a `ReplayGap` as a `{"replayGap": {...}}` row via a new `_event_row` helper instead of crashing in `MessageToDict`. CLI-36: the Go CLI branches on `result.IsReplayGap()` and prints the typed `REPLAY_GAP requested_after= oldest_available=` line / `replayGap` JSON row instead of formatting the library's cleared `Event`. `docs/CrossLanguageSmokeMatrix.md` gained a per-CLI gap-rendering table (one edit covering both client findings). Four new tests as designed (3 × `SessionEventDistributorTests`, `GatewayEndToEndReconnectReplayTests.ReconnectAfterFullAgeEvictionResumesWithSentinelFormula`) plus `test_stream_events_renders_replay_gap` (Python) and `TestRunStreamEventsPrintsReplayGap` (Go); all written red first and each reproducing its defect verbatim. **Deferred:** GWC-25's `ReplayGap.oldest_available_sequence` proto-comment amendment is **not** in this change — it is comment-only but triggers the full five-client regen fan-out, so it lands with the later codegen wave (alongside IPC-23's proto-comment edits) rather than forcing a regen for one sentence. Note for that wave: the fake-worker gateway e2e suite cannot run on the macOS worktree without `TMPDIR` shortened (macOS caps the Unix-domain-socket path backing .NET named pipes at 104 chars; `TMPDIR=/tmp dotnet test …` works and was used here). | diff --git a/archreview/2026-07-12/remediation/10-gateway-core.md b/archreview/2026-07-12/remediation/10-gateway-core.md index 7d98ccb..5f808ee 100644 --- a/archreview/2026-07-12/remediation/10-gateway-core.md +++ b/archreview/2026-07-12/remediation/10-gateway-core.md @@ -9,7 +9,7 @@ This document turns the 2026-07-12 re-review's **new** Gateway Server Core findi | ID | Sev | Tier | Eff | Dep | Status | Title | |----|-----|------|-----|-----|--------|-------| | GWC-24 | Medium | P1 | M | GWC-21 (coord) | Not started | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | -| GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | +| GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Done | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | | GWC-26 | Low | P2 | M | GWC-27 | Not started | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed, missed Acknowledge never repaired | | GWC-27 | Low | P2 | S | GWC-26 | Not started | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor permanently | | GWC-28 | Low | P2 | S | GWC-10 (coord) | Not started | Gateway→worker envelope `sequence` stamped at creation, not at write — non-monotonic on the wire under concurrent invokes | diff --git a/archreview/2026-07-12/remediation/50-clients.md b/archreview/2026-07-12/remediation/50-clients.md index c1965f0..41d0d34 100644 --- a/archreview/2026-07-12/remediation/50-clients.md +++ b/archreview/2026-07-12/remediation/50-clients.md @@ -16,8 +16,8 @@ Operating constraints carried from prior work: | ID | Sev | Tier | Eff | Dep | Status | Title | |----|-----|------|-----|-----|--------|-------| -| CLI-35 | Medium | P0 | S | — | Not started | Python CLI `stream-events` crashes on a ReplayGap | -| CLI-36 | Medium | P0 | S | — | Not started | Go CLI `stream-events` silently destroys the ReplayGap signal | +| CLI-35 | Medium | P0 | S | — | Done | Python CLI `stream-events` crashes on a ReplayGap | +| CLI-36 | Medium | P0 | S | — | Done | Go CLI `stream-events` silently destroys the ReplayGap signal | | CLI-37 | Medium | P1 | M | CLI-38 | Not started | Status-array validation must branch on `category` per the proto contract (4-vs-1 divergence) | | CLI-38 | Medium | P1 | S | — | Not started | 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 | diff --git a/clients/go/cmd/mxgw-go/main.go b/clients/go/cmd/mxgw-go/main.go index 3a25ade..2695363 100644 --- a/clients/go/cmd/mxgw-go/main.go +++ b/clients/go/cmd/mxgw-go/main.go @@ -57,6 +57,25 @@ type commandReplyOutput struct { Reply json.RawMessage `json:"reply"` } +// replayGapRow is the JSON row stream-events emits for a reconnect-replay gap: +// {"replayGap":{"requestedAfterSequence":N,"oldestAvailableSequence":N}}. +// +// The cursors are typed by hand rather than marshalled with protojson on +// purpose. The proto3 JSON mapping renders 64-bit integers as JSON *strings* +// ("7"), but the Rust and Python CLIs emit JSON *numbers* (7) for this row — +// routing through protojson would silently make Go the odd one out and break +// the cross-language smoke matrix's row comparison. encoding/json renders +// uint64 as a number, which is the canonical rendering here. +type replayGapRow struct { + ReplayGap replayGapCursors `json:"replayGap"` +} + +// replayGapCursors is the nested cursor object of replayGapRow. +type replayGapCursors struct { + RequestedAfterSequence uint64 `json:"requestedAfterSequence"` + OldestAvailableSequence uint64 `json:"oldestAvailableSequence"` +} + func main() { if err := runWithIO(context.Background(), os.Args[1:], os.Stdout, os.Stderr); err != nil { fmt.Fprintln(os.Stderr, err) @@ -970,7 +989,31 @@ func runStreamEvents(ctx context.Context, args []string, stdout, stderr io.Write if result.Err != nil { return result.Err } - if *jsonOutput { + // A reconnect-replay gap is a typed signal, not an event: the library + // clears Event on it, so formatting Event here would print a meaningless + // zero row and discard the resume cursors the operator needs. Render it + // as its own row (matching the Rust CLI) and count it toward -limit like + // any other emitted row. + if result.IsReplayGap() { + if *jsonOutput { + row, err := json.Marshal(replayGapRow{ + ReplayGap: replayGapCursors{ + RequestedAfterSequence: result.ReplayGap.GetRequestedAfterSequence(), + OldestAvailableSequence: result.ReplayGap.GetOldestAvailableSequence(), + }, + }) + if err != nil { + return err + } + fmt.Fprintln(stdout, string(row)) + } else { + fmt.Fprintf( + stdout, + "REPLAY_GAP requested_after=%d oldest_available=%d\n", + result.ReplayGap.GetRequestedAfterSequence(), + result.ReplayGap.GetOldestAvailableSequence()) + } + } else if *jsonOutput { fmt.Fprintln(stdout, string(mustMarshalProto(result.Event))) } else { fmt.Fprintf(stdout, "%d %s\n", result.Event.GetWorkerSequence(), result.Event.GetFamily()) diff --git a/clients/go/cmd/mxgw-go/main_test.go b/clients/go/cmd/mxgw-go/main_test.go index b3102d8..3c5a4d2 100644 --- a/clients/go/cmd/mxgw-go/main_test.go +++ b/clients/go/cmd/mxgw-go/main_test.go @@ -617,3 +617,120 @@ func TestRunWriteBulkVariantRejectsMismatchedHandlesAndValues(t *testing.T) { t.Fatalf("write-bulk mismatched handles/values error = %v", err) } } + +// replayGapFakeGateway streams the gateway's reconnect-replay sentinel (an MxEvent +// carrying replay_gap, family UNSPECIFIED, body unset) followed by one normal data +// event — exactly what a resume whose cursor predates the retained replay ring sees. +type replayGapFakeGateway struct { + pb.UnimplementedMxAccessGatewayServer +} + +func (g *replayGapFakeGateway) StreamEvents( + req *pb.StreamEventsRequest, + stream grpc.ServerStreamingServer[pb.MxEvent], +) error { + sentinel := &pb.MxEvent{ + SessionId: req.GetSessionId(), + Family: pb.MxEventFamily_MX_EVENT_FAMILY_UNSPECIFIED, + ReplayGap: &pb.ReplayGap{ + RequestedAfterSequence: 7, + OldestAvailableSequence: 42, + }, + } + if err := stream.Send(sentinel); err != nil { + return err + } + return stream.Send(&pb.MxEvent{ + SessionId: req.GetSessionId(), + Family: pb.MxEventFamily_MX_EVENT_FAMILY_ON_DATA_CHANGE, + WorkerSequence: 43, + }) +} + +func startReplayGapGateway(t *testing.T) string { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + server := grpc.NewServer() + pb.RegisterMxAccessGatewayServer(server, &replayGapFakeGateway{}) + go func() { _ = server.Serve(listener) }() + t.Cleanup(func() { + server.Stop() + _ = listener.Close() + }) + return listener.Addr().String() +} + +// TestRunStreamEventsPrintsReplayGap pins CLI-36: the CLI must render the typed +// ReplayGap signal in both output modes instead of formatting the library's +// cleared Event field (which printed "0 MX_EVENT_FAMILY_UNSPECIFIED" in text mode +// and an empty object in JSON mode, destroying the resume cursors). +func TestRunStreamEventsPrintsReplayGap(t *testing.T) { + endpoint := startReplayGapGateway(t) + + baseArgs := []string{ + "stream-events", + "-endpoint", endpoint, + "-plaintext", + "-api-key", "test", + "-session-id", "gap-session", + "-after-worker-sequence", "7", + "-limit", "2", + } + + var stdout, stderr bytes.Buffer + if err := runWithIO(t.Context(), baseArgs, &stdout, &stderr); err != nil { + t.Fatalf("runWithIO() error = %v; stderr = %s", err, stderr.String()) + } + text := stdout.String() + if !strings.Contains(text, "REPLAY_GAP requested_after=7 oldest_available=42") { + t.Fatalf("stream-events text output missing typed gap row: %q", text) + } + if strings.Contains(text, "0 MX_EVENT_FAMILY_UNSPECIFIED") { + t.Fatalf("stream-events text output destroyed the gap into a zero row: %q", text) + } + if !strings.Contains(text, "43 MX_EVENT_FAMILY_ON_DATA_CHANGE") { + t.Fatalf("stream-events text output dropped the normal event: %q", text) + } + + stdout.Reset() + stderr.Reset() + if err := runWithIO(t.Context(), append(baseArgs, "-json"), &stdout, &stderr); err != nil { + t.Fatalf("runWithIO(-json) error = %v; stderr = %s", err, stderr.String()) + } + + lines := strings.Split(strings.TrimSpace(stdout.String()), "\n") + if len(lines) != 2 { + t.Fatalf("stream-events -json emitted %d rows, want 2: %q", len(lines), stdout.String()) + } + + // The cursors must decode as JSON numbers, not the strings the proto3 JSON + // mapping would produce for 64-bit fields: the Rust and Python CLIs emit + // numbers, and the cross-language matrix compares these rows across clients. + var gapRow struct { + ReplayGap *struct { + RequestedAfterSequence uint64 `json:"requestedAfterSequence"` + OldestAvailableSequence uint64 `json:"oldestAvailableSequence"` + } `json:"replayGap"` + } + if err := json.Unmarshal([]byte(lines[0]), &gapRow); err != nil { + t.Fatalf("parse gap row: %v\nrow: %s", err, lines[0]) + } + if gapRow.ReplayGap == nil { + t.Fatalf("stream-events -json first row is not a replayGap row: %s", lines[0]) + } + if gapRow.ReplayGap.RequestedAfterSequence != 7 || gapRow.ReplayGap.OldestAvailableSequence != 42 { + t.Fatalf("stream-events -json gap cursors = %+v, want 7/42", *gapRow.ReplayGap) + } + // Belt and braces on the value type: a protojson-rendered `"7"` already + // fails the decode above (encoding/json rejects a JSON string for an + // untagged uint64 field), but assert the raw bytes so a regression names + // the real problem instead of surfacing as an opaque unmarshal error. + if !strings.Contains(lines[0], `"requestedAfterSequence":7`) || + !strings.Contains(lines[0], `"oldestAvailableSequence":42`) { + t.Fatalf("stream-events -json gap cursors must be JSON numbers, got: %s", lines[0]) + } +} diff --git a/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py b/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py index 154c339..65bf969 100644 --- a/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py +++ b/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py @@ -21,6 +21,7 @@ from zb_mom_ww_mxgateway import __version__ from zb_mom_ww_mxgateway.auth import redact_secret from zb_mom_ww_mxgateway.client import GatewayClient from zb_mom_ww_mxgateway.errors import MxGatewayError +from zb_mom_ww_mxgateway.events import ReplayGap from zb_mom_ww_mxgateway.galaxy import GalaxyRepositoryClient from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb @@ -1103,7 +1104,7 @@ async def _stream_events(**kwargs: Any) -> dict[str, Any]: max_events=kwargs["max_events"], timeout=kwargs["timeout"], ) - return {"events": [_message_dict(event) for event in events]} + return {"events": [_event_row(event) for event in events]} async def _stream_alarms(**kwargs: Any) -> dict[str, Any]: @@ -1500,14 +1501,14 @@ async def _collect_events( *, max_events: int, timeout: float, -) -> list[pb.MxEvent]: +) -> list[pb.MxEvent | ReplayGap]: if max_events > MAX_AGGREGATE_EVENTS: raise click.BadParameter( f"must be less than or equal to {MAX_AGGREGATE_EVENTS}", param_hint="--max-events", ) - collected: list[pb.MxEvent] = [] + collected: list[pb.MxEvent | ReplayGap] = [] iterator = events.__aiter__() try: while len(collected) < max_events: @@ -1630,3 +1631,26 @@ def _message_dict(message: Any) -> dict[str, Any]: preserving_proto_field_name=False, use_integers_for_enums=False, ) + + +def _event_row(item: Any) -> dict[str, Any]: + """Render one item of an event stream as a JSON row. + + ``Session.stream_events`` yields ``MxEvent | ReplayGap``. ``ReplayGap`` is a + plain dataclass, so it has no protobuf descriptor and cannot go through + ``MessageToDict`` — it gets its own distinct row instead, matching the shape + the Rust and Go CLIs emit so the cross-language matrix can compare rows. + Keys are camelCase for the same reason ``_message_dict`` uses + ``preserving_proto_field_name=False``. The gap is always rendered: never + dropped, and never re-synthesized into an event. + """ + + if isinstance(item, ReplayGap): + return { + "replayGap": { + "requestedAfterSequence": item.requested_after_sequence, + "oldestAvailableSequence": item.oldest_available_sequence, + }, + } + + return _message_dict(item) diff --git a/clients/python/tests/test_cli.py b/clients/python/tests/test_cli.py index 856b83f..0da08a4 100644 --- a/clients/python/tests/test_cli.py +++ b/clients/python/tests/test_cli.py @@ -817,3 +817,65 @@ def test_write_secured_command_does_not_echo_value_on_failure( def test_write_secured_and_authenticate_user_commands_are_registered() -> None: names = set(main.commands) assert {"write-secured", "authenticate-user"} <= names + + +class _FakeReplayGapSession: + """Session stand-in whose event stream starts with a ReplayGap sentinel. + + Mirrors what ``Session.stream_events`` yields on a resume that predates the + gateway's retained replay ring: the typed gap first, then normal events. + """ + + def __init__(self, gap, event) -> None: + self._gap = gap + self._event = event + + def stream_events(self, **_kwargs): + async def _iterate(): + yield self._gap + yield self._event + + return _iterate() + + +def test_stream_events_renders_replay_gap(monkeypatch: pytest.MonkeyPatch) -> None: + """CLI-35: a ReplayGap renders as its own JSON row instead of crashing the command.""" + from zb_mom_ww_mxgateway.events import ReplayGap + from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb + + gap = ReplayGap(requested_after_sequence=7, oldest_available_sequence=42) + event = pb.MxEvent(session_id="cli-test-session", worker_sequence=43) + + async def fake_connect(options, **_kwargs): + return _FakeAsyncClient() + + monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect) + monkeypatch.setattr( + commands_module, + "_session", + lambda _client, _session_id: _FakeReplayGapSession(gap, event), + ) + + result = CliRunner().invoke( + main, + [ + "stream-events", + "--plaintext", + "--session-id", + "cli-test-session", + "--after-worker-sequence", + "7", + "--max-events", + "2", + "--json", + ], + ) + + assert result.exit_code == 0, result.output + rows = json.loads(result.output)["events"] + assert rows[0] == { + "replayGap": {"requestedAfterSequence": 7, "oldestAvailableSequence": 42}, + } + # The gap is rendered, never swallowed, and the normal event still follows it. + assert "replayGap" not in rows[1] + assert rows[1]["workerSequence"] == "43" diff --git a/docs/CrossLanguageSmokeMatrix.md b/docs/CrossLanguageSmokeMatrix.md index af7c374..e0547b3 100644 --- a/docs/CrossLanguageSmokeMatrix.md +++ b/docs/CrossLanguageSmokeMatrix.md @@ -34,9 +34,40 @@ When `stream-events` is resumed with an `after_worker_sequence` cursor that predates the oldest event still in the gateway's replay ring, the gateway emits a single `ReplayGap` sentinel at the head of the stream. Every client surfaces this as a distinct, typed, non-terminal signal (see each client README); the resume -contract is `after_worker_sequence = oldest_available_sequence - 1`. The default -smoke sequence opens a fresh stream (no cursor) and does not exercise the gap -path; a resume-with-gap fixture case is tracked separately (TST-24). +contract is `after_worker_sequence = oldest_available_sequence - 1`, and it holds +even when the ring has been emptied entirely by age eviction — the gateway then +reports the next deliverable sequence rather than `0` (see [Sessions](Sessions.md)). +The default smoke sequence opens a fresh stream (no cursor) and does not exercise +the gap path; a resume-with-gap fixture case is tracked separately (TST-24). + +The CLIs differ in how they *print* that library-level signal. Three of them consume +the typed gap and emit a dedicated row rather than a degenerate event row; the other +two hand the raw sentinel `MxEvent` straight to the formatter, so they print the +sentinel itself, whose `replayGap` field carries the same cursors: + +| CLI | Text mode | JSON mode | +|-----|-----------|-----------| +| `mxgw-rs` (Rust, canonical) | `REPLAY_GAP requested_after= oldest_available=` | `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` as one entry of the `events` array | +| `mxgw-go` (Go) | `REPLAY_GAP requested_after= oldest_available=` | one `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` line, counted toward `-limit` like any other row | +| `mxgw-py` (Python) | same JSON dump as `--json` | `{"replayGap": {"requestedAfterSequence": , "oldestAvailableSequence": }}` as one entry of the `events` array | +| `mxgw-dotnet` (.NET) | the raw sentinel `MxEvent` as protobuf JSON, including its `replayGap` field | same, as one entry of the `events` array | +| `mxgw-java` (Java) | the sentinel's `worker_sequence` and `family` (`0 MX_EVENT_FAMILY_UNSPECIFIED`) | the raw sentinel `MxEvent` as protobuf JSON, including its `replayGap` field | + +Rust, Go, and Python emit the same two key names and, deliberately, the same JSON +value **types**: the cursors are JSON numbers (`7`), not strings. That is why the +Go CLI types the row by hand instead of marshalling `ReplayGap` with `protojson` — +the proto3 JSON mapping renders 64-bit integers as strings (`"7"`), which is also +why the .NET and Java rows, which pass the sentinel through a protobuf JSON +formatter, carry **quoted** cursors. A matrix runner must therefore compare parsed +values, not raw bytes, and must not assume the same value type across all five +CLIs. + +Two further formatting differences among the three canonical CLIs, none of them +semantic: Python sorts object keys and uses `", "` / `": "` separators +(`json.dumps(..., sort_keys=True)`), while Rust and Go emit compact, +declaration-ordered JSON; and the row sits alone on its own line for Go and for +Rust's `--jsonl`, but inside an `events` array for Python and for Rust's +aggregate `--json`. ## Integration Gate diff --git a/docs/Sessions.md b/docs/Sessions.md index 77a9db1..f8f56c1 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -225,12 +225,14 @@ The handoff is sealed by a watermark. `RegisterWithReplay` returns `LiveResumeSe Emit order on a resumed stream: -1. **ReplayGap sentinel (only when events were evicted).** If the requested `after_worker_sequence` predates the oldest event still retained — i.e. events in the open interval were dropped by capacity or age eviction and are unrecoverable — the gateway first yields a single sentinel `MxEvent` with `replay_gap` populated (`requested_after_sequence` = the requested watermark, `oldest_available_sequence` = the oldest still-retained sequence). The sentinel carries the session id; its `family` is `UNSPECIFIED`, its `body` oneof is unset, and no per-item fields are populated. It is an explicit, documented control signal — *not* a synthesized MXAccess event — telling the client to discard local state and re-snapshot. A client that wants to resume without another gap should set `after_worker_sequence = oldest_available_sequence - 1` on its next request. +1. **ReplayGap sentinel (only when events were evicted).** If the requested `after_worker_sequence` predates the oldest event still retained — i.e. events in the open interval were dropped by capacity or age eviction and are unrecoverable — the gateway first yields a single sentinel `MxEvent` with `replay_gap` populated (`requested_after_sequence` = the requested watermark, `oldest_available_sequence` = the resume anchor described below). The sentinel carries the session id; its `family` is `UNSPECIFIED`, its `body` oneof is unset, and no per-item fields are populated. It is an explicit, documented control signal — *not* a synthesized MXAccess event — telling the client to discard local state and re-snapshot. A client that wants to resume without another gap should set `after_worker_sequence = oldest_available_sequence - 1` on its next request. 2. **Retained replay batch.** The still-retained events newer than the requested watermark, in ascending `worker_sequence` order. 3. **Live events**, resuming strictly after `LiveResumeSequence`. When `after_worker_sequence` is inside the retained window (nothing was evicted), step 1 is skipped: the stream replays the retained tail then resumes live with no sentinel. +**`oldest_available_sequence` when the ring is empty.** Age eviction (`ReplayRetentionSeconds`, default 300) and a disabled ring (`ReplayBufferCapacity = 0`) both leave nothing retained, so there is no oldest-retained sequence to report. In that case the sentinel carries the **next sequence that can possibly be delivered** — the highest sequence the distributor has observed plus one — rather than `0`. That keeps `after_worker_sequence = oldest_available_sequence - 1` the single universal resume formula: the follow-up resume lands exactly on the highest observed sequence, replays nothing, reports no gap, and receives every subsequent live event. Reporting `0` here would make an unsigned client compute `2^64 - 1` and then silently receive nothing, because the live filter drops every event at or below that watermark. Nothing is lost relative to reporting `0` — the evicted interval is unrecoverable either way, and the sentinel's job is to tell the client to re-snapshot. `0` remains the value when there is no gap, where the field is meaningless and never emitted. + The ReplayGap sentinel is emitted **only** on the `StreamEvents` server stream and only to the resuming subscriber — it is never fanned to other subscribers and never appears in `DrainEventsReply` (the diagnostic drain path is untouched). Replay retention itself is bounded by `MxGateway:Events:ReplayBufferCapacity` (count) and `ReplayRetentionSeconds` (age); see [Configuration](GatewayConfiguration.md). ### Close diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs index 29e910a..797122b 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs @@ -390,10 +390,15 @@ public sealed class SessionEventDistributor : IAsyncDisposable /// gap semantics. /// /// - /// The oldest worker sequence still retained and replayable. 0 when nothing is - /// retained. Meaningful to the caller only when is - /// (it populates the ReplayGap sentinel's - /// oldest_available_sequence). + /// The resume anchor reported to a gapped client: the oldest worker sequence still + /// retained and replayable, or — when age/capacity eviction has emptied the ring + /// entirely — the next sequence that can possibly be delivered (highest observed + 1). + /// Either way the client's documented + /// after_worker_sequence = oldest_available_sequence - 1 formula yields a cursor + /// that resumes without dropping live events. 0 when is + /// , where the value is meaningless and never emitted. Meaningful + /// to the caller only when is (it populates + /// the ReplayGap sentinel's oldest_available_sequence). /// /// /// The worker sequence the live channel must resume strictly after: the highest @@ -463,7 +468,20 @@ public sealed class SessionEventDistributor : IAsyncDisposable if (_replayCount == 0) { gap = _anyEventSeen && afterSequence < _highestSequenceSeen; - oldestAvailableSequence = 0; // meaningful only when gap == true; 0 here since nothing is retained + + // GWC-25: nothing is retained, but a gapped client still needs a usable resume + // anchor. The documented client formula is + // after_worker_sequence = oldest_available_sequence - 1, so reporting 0 here made + // an unsigned client compute ulong.MaxValue: the follow-up resume then replayed + // nothing and reported no gap (MaxValue is below no real sequence, in this branch + // and in the retained branch's wrap guard alike), and the caller's live filter + // (sequence > liveResumeSequence) dropped every subsequent event — a silently + // dead stream. Reporting the next sequence that can possibly + // be delivered (highest observed + 1) makes oldest - 1 land exactly on the + // highest observed sequence, so the resume delivers everything newer. Nothing is + // recoverable either way; the sentinel's job is to say "re-snapshot". + // Still 0 when gap == false, where the field is documented as meaningless. + oldestAvailableSequence = gap ? _highestSequenceSeen + 1 : 0; } else { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs index 424bd66..7bfa802 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs @@ -2,6 +2,7 @@ using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; +using Microsoft.Extensions.Time.Testing; using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Server.Configuration; @@ -231,6 +232,114 @@ public sealed class GatewayEndToEndReconnectReplayTests Assert.Equal(expectedTail, tail.Select(e => e.WorkerSequence).ToArray()); } + /// + /// GWC-25 regression, end to end: when age eviction has emptied the replay ring, the + /// ReplayGap sentinel carries the next deliverable sequence (highest observed + 1), + /// so a client applying the documented + /// after_worker_sequence = oldest_available_sequence - 1 formula on its follow-up + /// resume receives subsequent live events. Under the pre-fix sentinel value of 0 + /// the formula wrapped to and the resumed stream never + /// delivered another event. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReconnectAfterFullAgeEvictionResumesWithSentinelFormula() + { + const int firstBatch = 4; + const double retentionSeconds = 30; + + FakeTimeProvider clock = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + GatedEventFakeWorkerProcessLauncher launcher = new(); + + // Capacity is ample; age is the eviction axis under test. + await using ReconnectReplayGatewayServiceFixture fixture = new( + launcher, + replayBufferCapacity: 16, + replayRetentionSeconds: retentionSeconds, + timeProvider: clock); + + string sessionId = await OpenSessionAsync(fixture, "reconnect-aged-out"); + + using CancellationTokenSource writer1Cts = new(); + RecordingServerStreamWriter writer1 = new(); + Task stream1Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId }, + writer1, + new TestServerCallContext(cancellationToken: writer1Cts.Token))); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + await WireUpAdviseAsync(fixture, sessionId); + + for (int i = 0; i < firstBatch; i++) + { + launcher.AllowNextEvent(); + } + + IReadOnlyList batch1 = await writer1.WaitForMessageCountAsync(firstBatch, TestTimeout); + ulong[] batch1Sequences = batch1.Select(e => e.WorkerSequence).ToArray(); + + // The client's cursor is mid-batch: it detached before consuming the tail, so it is + // genuinely behind the highest sequence the distributor observed. + ulong staleCursor = batch1Sequences[1]; + ulong highestSeen = batch1Sequences[firstBatch - 1]; + + await DetachAsync(writer1Cts, stream1Task); + + // Age every retained event out of the ring: the next resume finds it empty. + clock.Advance(TimeSpan.FromSeconds(retentionSeconds * 2)); + + // ---- first reconnect: receives only the sentinel, nothing is replayable ---- + using CancellationTokenSource writer2Cts = new(); + RecordingServerStreamWriter writer2 = new(); + Task stream2Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = staleCursor }, + writer2, + new TestServerCallContext(cancellationToken: writer2Cts.Token))); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + IReadOnlyList gapOnly = await writer2.WaitForMessageCountAsync(1, TestTimeout); + + MxEvent sentinel = gapOnly[0]; + Assert.NotNull(sentinel.ReplayGap); + Assert.Equal(staleCursor, sentinel.ReplayGap.RequestedAfterSequence); + + // The whole point of GWC-25: the sentinel anchors the resume at the next sequence that + // can still be delivered, not at 0. + Assert.Equal(highestSeen + 1, sentinel.ReplayGap.OldestAvailableSequence); + + await DetachAsync(writer2Cts, stream2Task); + + // ---- second reconnect: the client applies the documented oldest - 1 formula ---- + // Unchecked ulong arithmetic: with the pre-fix sentinel value of 0 this wraps to + // ulong.MaxValue and the live filter drops every subsequent event. + ulong resumeCursor = sentinel.ReplayGap.OldestAvailableSequence - 1; + + RecordingServerStreamWriter writer3 = new(); + Task stream3Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = resumeCursor }, + writer3, + new TestServerCallContext())); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + + launcher.AllowNextEvent(); + IReadOnlyList live = await writer3.WaitForMessageCountAsync(1, TestTimeout); + + launcher.StopEmitting(); + await CloseAndDrainAsync(fixture, sessionId, stream3Task, launcher); + + // A live event arrives on the resumed stream, and it is a real event, not another gap. + MxEvent delivered = live[0]; + Assert.Null(delivered.ReplayGap); + Assert.Equal(MxEventFamily.OnDataChange, delivered.Family); + Assert.True( + delivered.WorkerSequence > highestSeen, + $"Live event {delivered.WorkerSequence} must be newer than the highest previously seen {highestSeen}."); + } + // ---- shared flow helpers ---- private static async Task OpenSessionAsync( @@ -385,11 +494,23 @@ public sealed class GatewayEndToEndReconnectReplayTests /// Initializes a new instance of the class. /// Fake worker process launcher backing the session manager. /// Replay ring capacity for the session's event distributor. + /// + /// Replay retention window. The default keeps age-eviction effectively off for the + /// duration of a fast test; the age-eviction test shortens it and drives + /// . + /// + /// + /// Clock handed to the session manager, which flows through to the session's event + /// distributor. Pass a fake to make age-eviction deterministic. + /// public ReconnectReplayGatewayServiceFixture( IWorkerProcessLauncher launcher, - int replayBufferCapacity) + int replayBufferCapacity, + double replayRetentionSeconds = 300, + TimeProvider? timeProvider = null) { - IOptions options = Options.Create(CreateOptions(replayBufferCapacity)); + IOptions options = Options.Create( + CreateOptions(replayBufferCapacity, replayRetentionSeconds)); SessionWorkerClientFactory workerClientFactory = new( launcher, options, @@ -401,6 +522,7 @@ public sealed class GatewayEndToEndReconnectReplayTests options, _metrics, logger: NullLogger.Instance, + timeProvider: timeProvider, dashboardEventBroadcaster: NullDashboardEventBroadcaster.Instance); MxAccessGrpcMapper mapper = new(); EventStreamService eventStreamService = new( @@ -472,7 +594,7 @@ public sealed class GatewayEndToEndReconnectReplayTests _metrics.Dispose(); } - private static GatewayOptions CreateOptions(int replayBufferCapacity) => + private static GatewayOptions CreateOptions(int replayBufferCapacity, double replayRetentionSeconds) => new() { Worker = new WorkerOptions @@ -501,9 +623,10 @@ public sealed class GatewayEndToEndReconnectReplayTests QueueCapacity = 32, ReplayBufferCapacity = replayBufferCapacity, - // Keep age-eviction effectively off for the duration of a fast test so - // capacity is the only eviction axis under test. - ReplayRetentionSeconds = 300, + // Defaults to a window long enough that age-eviction never fires during a + // fast test, so capacity is the only eviction axis; the age-eviction test + // shortens it and drives a fake clock. + ReplayRetentionSeconds = replayRetentionSeconds, }, }; } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs index 75a2544..e910549 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs @@ -841,6 +841,153 @@ public sealed class SessionEventDistributorTests Assert.Equal(4ul, live.WorkerSequence); } + /// + /// GWC-25: when age eviction has emptied the replay ring, a resume behind the highest + /// observed sequence still reports a gap, and the reported oldest-available sequence is + /// the next sequence that can possibly be delivered (highest seen + 1) — never 0, + /// which would make the client's documented oldest - 1 resume formula wrap to + /// and dead-stream the subscriber. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RegisterWithReplayReportsNextDeliverableSequenceWhenRingEmptiedByAge() + { + FakeTimeProvider time = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + Channel source = Channel.CreateUnbounded(); + await using SessionEventDistributor distributor = CreateDistributor( + source.Reader, + replayBufferCapacity: 100, + replayRetentionSeconds: 30, + timeProvider: time); + await distributor.StartAsync(CancellationToken.None); + + // A primer subscriber forces the pump to retain events 1..3 deterministically. + using IEventSubscriberLease primer = distributor.Register(); + for (ulong sequence = 1; sequence <= 3; sequence++) + { + source.Writer.TryWrite(Event(sequence)); + _ = await ReadOneAsync(primer.Reader); + } + + // Past the retention window: RegisterWithReplay's EvictAged() empties the ring entirely. + time.Advance(TimeSpan.FromSeconds(60)); + + using IEventSubscriberLease resume = distributor.RegisterWithReplay( + 1, + out IReadOnlyList replay, + out bool gap, + out ulong oldestAvailable, + out ulong liveResume); + + Assert.True(gap); + Assert.Empty(replay); + Assert.Equal(1ul, liveResume); + + // Highest seen is 3, so the next deliverable sequence is 4. + Assert.Equal(4ul, oldestAvailable); + } + + /// + /// GWC-25: the same next-deliverable-sequence rule applies when retention is disabled + /// outright (replay capacity 0) — the ring is always empty, so a behind cursor must still + /// get a usable resume anchor rather than 0. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RegisterWithReplayReportsNextDeliverableSequenceWithRetentionDisabled() + { + Channel source = Channel.CreateUnbounded(); + await using SessionEventDistributor distributor = CreateDistributor( + source.Reader, + replayBufferCapacity: 0, + replayRetentionSeconds: 0); + await distributor.StartAsync(CancellationToken.None); + + using IEventSubscriberLease primer = distributor.Register(); + for (ulong sequence = 1; sequence <= 3; sequence++) + { + source.Writer.TryWrite(Event(sequence)); + _ = await ReadOneAsync(primer.Reader); + } + + using IEventSubscriberLease resume = distributor.RegisterWithReplay( + 1, + out IReadOnlyList replay, + out bool gap, + out ulong oldestAvailable, + out _); + + Assert.True(gap); + Assert.Empty(replay); + Assert.Equal(4ul, oldestAvailable); + } + + /// + /// GWC-25 regression: a client that applies the documented + /// after_worker_sequence = oldest_available_sequence - 1 formula to an empty-ring + /// ReplayGap resumes with a live watermark equal to the highest observed sequence, so the + /// caller's live filter passes every subsequent event. Under the pre-fix 0 the + /// formula wrapped to and the stream went permanently silent. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ResumeUsingSentinelFormulaAfterEmptyRingGapDeliversLiveEvents() + { + FakeTimeProvider time = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + Channel source = Channel.CreateUnbounded(); + await using SessionEventDistributor distributor = CreateDistributor( + source.Reader, + replayBufferCapacity: 100, + replayRetentionSeconds: 30, + timeProvider: time); + await distributor.StartAsync(CancellationToken.None); + + using IEventSubscriberLease primer = distributor.Register(); + for (ulong sequence = 1; sequence <= 3; sequence++) + { + source.Writer.TryWrite(Event(sequence)); + _ = await ReadOneAsync(primer.Reader); + } + + time.Advance(TimeSpan.FromSeconds(60)); + + // First resume: everything is evicted, so the sentinel carries the next deliverable + // sequence. The subscriber is disposed immediately — the client reconnects with the + // formula below. + IEventSubscriberLease gapped = distributor.RegisterWithReplay( + 1, + out _, + out bool gap, + out ulong oldestAvailable, + out _); + gapped.Dispose(); + + Assert.True(gap); + + // The documented client formula. Unchecked ulong arithmetic: under the pre-fix value of + // 0 this wraps to ulong.MaxValue, which is exactly the dead-stream defect. + ulong resumeCursor = oldestAvailable - 1; + + using IEventSubscriberLease resumed = distributor.RegisterWithReplay( + resumeCursor, + out IReadOnlyList replay, + out bool resumedGap, + out _, + out ulong liveResume); + + Assert.False(resumedGap); + Assert.Empty(replay); + + // The live filter the caller applies is "sequence > liveResume": it must sit at the + // highest observed sequence so the next event passes. + Assert.Equal(3ul, liveResume); + + source.Writer.TryWrite(Event(4)); + MxEvent live = await ReadOneAsync(resumed.Reader); + Assert.Equal(4ul, live.WorkerSequence); + Assert.True(live.WorkerSequence > liveResume); + } + private static async Task DrainUntilFaultAsync(ChannelReader reader) { // Drains any buffered events, then surfaces the channel's completion fault (if any)