Merge branch 'fix/gwc-25-replaygap-trio'
ci / nightly-windev (push) Has been skipped
ci / java (push) Successful in 2m10s
ci / windows-x86 (push) Successful in 1m30s
ci / portable (push) Successful in 7m39s

# Conflicts:
#	archreview/2026-07-12/remediation/00-tracking.md
#	archreview/2026-07-12/remediation/10-gateway-core.md
This commit is contained in:
Joseph Doherty
2026-08-07 05:49:13 -04:00
12 changed files with 596 additions and 28 deletions
@@ -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) | Done | 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 |
@@ -163,3 +163,4 @@ Sequence these together rather than piecemeal — several are one change set spa
| 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 | **TST-29 → `Done`:** migrated the Phase-5 (orphan-worker reattach) deferred-not-planned governance record and the settled Phase-4 Viewer-default decision from `oldtasks.md` into a new "Session-Resilience Epic Scope" entry in `docs/DesignDecisions.md`; repointed CLAUDE.md and `stillpending.md:7,165` from `oldtasks.md` to `docs/DesignDecisions.md` / `docs/plans/2026-06-15-session-resilience.md.tasks.json`; `git rm oldtasks.md`. The five untracked root docs-review artifacts (`MxAccessGateway-docs-{issues,fixed,final}.md`, `MxGatewayClient-docs-{issues,fixed}.md`) were absent from this worktree — delete from the main working tree separately. |
| 2026-08-07 | **GWC-24 → `Done`** (branch `fix/gwc-24-staging-bound`). `WorkerClient._eventStaging` is now `Channel.CreateBounded` at `2 × EventChannelCapacity` (`Wait`, single reader/writer, no sync continuations); a rejected staging `TryWrite` faults the client `ProtocolViolation` with `QueueOverflow("worker-event-staging")` unless `IsTerminalState()` (shutdown stays a silent drop), so a consumer draining slower than its worker produces dies at a fixed ceiling instead of growing gateway memory. Queue-depth accounting moved from `EnqueueWorkerEventAsync` to `StageWorkerEvent`, so the single gauge reports staged + queued; the timed-write fault (`EventChannelFullModeTimeout` / `QueueOverflow("worker-events")`) is unchanged and still catches the full-stall case first. No new config key — total gateway-side buffering is `3 × MxGateway:Events:QueueCapacity`, derived; coordination with still-open old **GWC-21** (`EventChannelFullModeTimeout` configurability) remains open and was not blocked on. Docs same commit: `GatewayProcessDesign.md` (two overflow faults), `MxAccessWorkerInstanceDesign.md`, `GatewayConfiguration.md`, `Metrics.md`. Tests: `WorkerClientTests.StagingChannelOverflowFaultsWorkerWithoutWaitingForFullModeTimeout` and `.WorkerEventQueueDepthGaugeCountsStagedEvents`; `WorkerClientTests` 22/22 green, `NonWindows.slnx` builds with 0 warnings. |
| 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=<n> oldest_available=<n>` 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). |
@@ -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) | Done | 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 |
@@ -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 |
+43
View File
@@ -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
}
// 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())
+117
View File
@@ -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])
}
}
@@ -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)
+62
View File
@@ -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"
+34 -3
View File
@@ -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=<n> oldest_available=<n>` | `{"replayGap": {"requestedAfterSequence": <n>, "oldestAvailableSequence": <n>}}` as one entry of the `events` array |
| `mxgw-go` (Go) | `REPLAY_GAP requested_after=<n> oldest_available=<n>` | one `{"replayGap": {"requestedAfterSequence": <n>, "oldestAvailableSequence": <n>}}` line, counted toward `-limit` like any other row |
| `mxgw-py` (Python) | same JSON dump as `--json` | `{"replayGap": {"requestedAfterSequence": <n>, "oldestAvailableSequence": <n>}}` 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
+3 -1
View File
@@ -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
@@ -390,10 +390,15 @@ public sealed class SessionEventDistributor : IAsyncDisposable
/// <see cref="TryGetReplayFrom"/> gap semantics.
/// </param>
/// <param name="oldestAvailableSequence">
/// The oldest worker sequence still retained and replayable. <c>0</c> when nothing is
/// retained. Meaningful to the caller only when <paramref name="gap"/> is
/// <see langword="true"/> (it populates the ReplayGap sentinel's
/// <c>oldest_available_sequence</c>).
/// 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
/// <c>after_worker_sequence = oldest_available_sequence - 1</c> formula yields a cursor
/// that resumes without dropping live events. <c>0</c> when <paramref name="gap"/> is
/// <see langword="false"/>, where the value is meaningless and never emitted. Meaningful
/// to the caller only when <paramref name="gap"/> is <see langword="true"/> (it populates
/// the ReplayGap sentinel's <c>oldest_available_sequence</c>).
/// </param>
/// <param name="liveResumeSequence">
/// 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
{
@@ -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());
}
/// <summary>
/// GWC-25 regression, end to end: when age eviction has emptied the replay ring, the
/// <c>ReplayGap</c> sentinel carries the next deliverable sequence (highest observed + 1),
/// so a client applying the documented
/// <c>after_worker_sequence = oldest_available_sequence - 1</c> formula on its follow-up
/// resume receives subsequent live events. Under the pre-fix sentinel value of <c>0</c>
/// the formula wrapped to <see cref="ulong.MaxValue"/> and the resumed stream never
/// delivered another event.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<string> OpenSessionAsync(
@@ -385,11 +494,23 @@ public sealed class GatewayEndToEndReconnectReplayTests
/// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
/// <param name="launcher">Fake worker process launcher backing the session manager.</param>
/// <param name="replayBufferCapacity">Replay ring capacity for the session's event distributor.</param>
/// <param name="replayRetentionSeconds">
/// 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
/// <paramref name="timeProvider"/>.
/// </param>
/// <param name="timeProvider">
/// Clock handed to the session manager, which flows through to the session's event
/// distributor. Pass a fake to make age-eviction deterministic.
/// </param>
public ReconnectReplayGatewayServiceFixture(
IWorkerProcessLauncher launcher,
int replayBufferCapacity)
int replayBufferCapacity,
double replayRetentionSeconds = 300,
TimeProvider? timeProvider = null)
{
IOptions<GatewayOptions> options = Options.Create(CreateOptions(replayBufferCapacity));
IOptions<GatewayOptions> options = Options.Create(
CreateOptions(replayBufferCapacity, replayRetentionSeconds));
SessionWorkerClientFactory workerClientFactory = new(
launcher,
options,
@@ -401,6 +522,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
options,
_metrics,
logger: NullLogger<SessionManager>.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,
},
};
}
@@ -841,6 +841,153 @@ public sealed class SessionEventDistributorTests
Assert.Equal(4ul, live.WorkerSequence);
}
/// <summary>
/// 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 <c>0</c>,
/// which would make the client's documented <c>oldest - 1</c> resume formula wrap to
/// <see cref="ulong.MaxValue"/> and dead-stream the subscriber.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task RegisterWithReplayReportsNextDeliverableSequenceWhenRingEmptiedByAge()
{
FakeTimeProvider time = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
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<MxEvent> 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);
}
/// <summary>
/// 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 <c>0</c>.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task RegisterWithReplayReportsNextDeliverableSequenceWithRetentionDisabled()
{
Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
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<MxEvent> replay,
out bool gap,
out ulong oldestAvailable,
out _);
Assert.True(gap);
Assert.Empty(replay);
Assert.Equal(4ul, oldestAvailable);
}
/// <summary>
/// GWC-25 regression: a client that applies the documented
/// <c>after_worker_sequence = oldest_available_sequence - 1</c> 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 <c>0</c> the
/// formula wrapped to <see cref="ulong.MaxValue"/> and the stream went permanently silent.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task ResumeUsingSentinelFormulaAfterEmptyRingGapDeliversLiveEvents()
{
FakeTimeProvider time = new(new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
Channel<MxEvent> source = Channel.CreateUnbounded<MxEvent>();
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<MxEvent> 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<MxEvent> reader)
{
// Drains any buffered events, then surfaces the channel's completion fault (if any)