From 7dbdb25a0c9b6ad24227d21eb8090d6c5787ce26 Mon Sep 17 00:00:00 2001 From: htjulia Date: Mon, 21 Sep 2026 11:35:42 +0900 Subject: [PATCH] feat(web): ack a console stream connection with a ready frame On a platform with no traffic, "the stream is open" and "the server is actually running" are indistinguishable to a console client: nothing arrives on /platform/traces/stream until a request does, so an open socket to a dead ticker looks exactly like an open socket to a healthy idle one. The stream now answers the connection before it answers with data. The ready frame goes out first, ahead of the initial trace replay, and it repeats on every EventSource reconnect so a client re-learns the connect-time facts rather than carrying stale ones across the gap. It is an ack, not a data frame, so it carries only what a client must know once at connect time and cannot derive from the stream afterwards: platform/platformVersion which implementation and which version is on the other end. Three implementations share this contract, so a bare version number would leave the console guessing whose it is. The version comes from the jar manifest and is null without one -- a placeholder would be a fabricated answer to "which version is this". tracesEnabled separates "recording is off" from "recording metricsEnabled is on but idle". An empty stream does not. buffered the rows the following trace frame carries, not the rows the ring holds. The two diverge once trace.capacity passes the 200-row replay cap, and announcing a count while withholding the rows is worse than announcing neither. tickMs the push period, so a silence watchdog sizes itself against a contract, not an assumption. capacity trace.capacity as of this connection, so a client can say what is no longer on the server. It is a live key, hence "as of". Every value is read at connect time because all four config keys are live; one cached at startup would be a stale answer to a question the client only gets to ask once. Existing clients are unaffected -- EventSource drops named events that have no listener -- and later versions may add fields, which clients must ignore. Co-Authored-By: htjulia --- CHANGELOG.ko.md | 20 +++++- CHANGELOG.md | 25 +++++-- README.ko.md | 2 +- README.md | 2 +- docs/guide/04-rest-api.ko.md | 22 +++++- docs/guide/04-rest-api.md | 22 +++++- .../org/htcom/protean/web/StreamReady.java | 63 +++++++++++++++++ .../htcom/protean/web/TraceStreamService.java | 29 ++++++-- .../org/htcom/protean/TraceStreamE2ETest.java | 69 ++++++++++++++----- 9 files changed, 218 insertions(+), 36 deletions(-) create mode 100644 src/main/java/org/htcom/protean/web/StreamReady.java diff --git a/CHANGELOG.ko.md b/CHANGELOG.ko.md index 69f8be0..631f010 100644 --- a/CHANGELOG.ko.md +++ b/CHANGELOG.ko.md @@ -10,9 +10,23 @@ ## [Unreleased] -이번 릴리스에서 공개 API 는 추가·제거·변경이 **없다** — `0.0.1` 로 컴파일한 소비자는 그대로 -링크된다. 바뀐 것은 내장 MCP 툴이 **무엇을 답하는가**다. 이 툴들을 wrap·subclass·위임해서 -쓰고 있다면 이 절 끝의 마이그레이션 노트를 읽어야 한다. +콘솔 SSE 스트림의 wire 계약에 하나가 추가됐다(아래). 소비자가 컴파일하는 Java API 자체는 +`0.0.1` 에서 바뀌지 않았고 그대로 링크된다. 그 밖에 바뀐 것은 내장 MCP 툴이 **무엇을 답하는가**다. +이 툴들을 wrap·subclass·위임해서 쓰고 있다면 이 절 끝의 마이그레이션 노트를 읽어야 한다. + +### 추가 + +- `GET /platform/traces/stream` 이 초기 스냅샷보다 앞서 **`ready` 확인(ack) 프레임**을 보낸다. + 트래픽이 없는 플랫폼에서는 *"스트림이 열렸다"* 와 *"서버가 실제로 돌고 있다"* 가 클라이언트 + 눈에 구별되지 않는다 — 요청이 들어오기 전까지 아무것도 도착하지 않기 때문이다. 이 ack 가 그 + 간극을 메우며, 클라이언트가 연결 시점에만 알 수 있는 값을 싣는다: `platform` 과 + `platformVersion`(후자는 jar manifest 가 없는 레이아웃에서 임의의 대체 문자열 대신 `null`), + `tracesEnabled` / `metricsEnabled`("기록이 꺼져 있다" 와 "켜져 있는데 한가하다" 를 가른다 — 빈 + 스트림으로는 안 된다), `buffered`(**뒤따르는 `trace` 프레임이 싣는 행 수**이며 링이 보유한 행 + 수가 아니다), `tickMs`(침묵 감지 워치독이 가정 대신 계약을 기준으로 삼게 한다), 그리고 이 + 연결 시점의 `capacity`. `EventSource` 가 재연결할 때마다 다시 나가며, 클라이언트는 모르는 + 필드를 무시해야 한다. 기존 클라이언트에는 영향이 없다 — `EventSource` 는 리스너가 없는 named + 이벤트를 버린다. 문서는 `docs/guide/04-rest-api.ko.md`. ### 수정 diff --git a/CHANGELOG.md b/CHANGELOG.md index b734e2c..b03f597 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,10 +10,27 @@ While the version is `0.x`, the public API may change between minor releases. ## [Unreleased] -No public API was added, removed, or changed in this release: a consumer compiled -against `0.0.1` links unchanged. What changed is what the built-in MCP tools -*answer*. If you wrap, subclass, or delegate to them, read the migration note at -the end of this section. +One addition to the wire contract of the console SSE stream (below); the Java API a +consumer compiles against is otherwise unchanged from `0.0.1` and still links. What +else changed is what the built-in MCP tools *answer*. If you wrap, subclass, or +delegate to them, read the migration note at the end of this section. + +### Added + +- `GET /platform/traces/stream` now opens with a **`ready` acknowledgement frame**, + sent ahead of the initial snapshot. On a platform with no traffic, *"the stream is + open"* and *"the server is actually running"* are indistinguishable to a client — + nothing arrives until a request does. The ack closes that gap and carries what a + client can only learn at connect time: `platform` and `platformVersion` (the latter + `null` when running from a layout with no jar manifest, rather than a fabricated + placeholder), `tracesEnabled` / `metricsEnabled` (which separate "recording is off" + from "recording is on but idle" — an empty stream does not), `buffered` (**the rows + the following `trace` frame carries**, not the rows the ring holds), `tickMs` (so a + silence watchdog has a contract to size itself against instead of an assumption), + and `capacity` as of this connection. It repeats on every `EventSource` reconnect, + and clients must ignore fields they do not recognize. Existing clients are + unaffected — `EventSource` drops named events that have no listener. Documented in + `docs/guide/04-rest-api.md`. ### Fixed diff --git a/README.ko.md b/README.ko.md index 198863f..1dec4f9 100644 --- a/README.ko.md +++ b/README.ko.md @@ -164,7 +164,7 @@ curl -X POST localhost:8080/platform/modules \ |--------|------|------| | `GET` | `/platform/traces` | 요청 트레이스(최신순). 필터 `moduleId`·`errorsOnly`·`status`·`minLatencyMs`·`since`·`beforeSeq`·`limit` | | `GET` | `/platform/traces/metrics` | 모듈별 집계 메트릭(요청수·에러율·지연 p50/p95/p99). `protean.trace.metrics.enabled` 필요 | -| `GET` | `/platform/traces/stream` | 라이브 SSE 푸시(`trace`/`metrics`/`modules`/`summary` 이벤트) | +| `GET` | `/platform/traces/stream` | 라이브 SSE 푸시(`ready` ack 후 `trace`/`metrics`/`modules`/`summary` 이벤트) | ### 설정 REST — `/platform/config` (`ConfigAdminController`, `protean.admin.enabled`) | 메서드 | 경로 | 동작 | diff --git a/README.md b/README.md index d9e3f2b..342cec9 100644 --- a/README.md +++ b/README.md @@ -202,7 +202,7 @@ curl -X POST localhost:8080/platform/modules \ |--------|------|------| | `GET` | `/platform/traces` | Request traces (newest-first). Filters `moduleId`·`errorsOnly`·`status`·`minLatencyMs`·`since`·`beforeSeq`·`limit` | | `GET` | `/platform/traces/metrics` | Per-module aggregate metrics (count, error rate, latency p50/p95/p99). Requires `protean.trace.metrics.enabled` | -| `GET` | `/platform/traces/stream` | Live SSE push (`trace`/`metrics`/`modules`/`summary` events) | +| `GET` | `/platform/traces/stream` | Live SSE push (`ready` ack, then `trace`/`metrics`/`modules`/`summary` events) | ### Config REST — `/platform/config` (`ConfigAdminController`, `protean.admin.enabled`) | Method | Path | Action | diff --git a/docs/guide/04-rest-api.ko.md b/docs/guide/04-rest-api.ko.md index bf460fe..7048659 100644 --- a/docs/guide/04-rest-api.ko.md +++ b/docs/guide/04-rest-api.ko.md @@ -393,16 +393,36 @@ POST /platform/modules/cp-mod/approve?approver=alice ### GET `/platform/traces/stream` -라이브 푸시 스트림(SSE) — 콘솔이 5초 폴링 대신 쓴다. 한 연결이 네 종류의 named 이벤트를 멀티플렉싱하며, 새로 연 연결은 네 종류의 초기 스냅샷을 먼저 받은 뒤 증분 푸시(대략 1초마다)를 받는다. +라이브 푸시 스트림(SSE) — 콘솔이 5초 폴링 대신 쓴다. 연결은 `ready` 프레임으로 확인(ack)된 뒤 네 종류의 named 이벤트를 멀티플렉싱하며, 새로 연 연결은 네 종류의 초기 스냅샷을 먼저 받은 뒤 증분 푸시(대략 1초마다)를 받는다. - Produces: `text/event-stream` - named 이벤트: + - `ready` — 연결 확인(ack). 언제나 첫 프레임이다(아래 참고) - `trace` — 새 `RequestTrace` 델타 - `metrics` — `ModuleMetricsSnapshot[]` 갱신(모듈별 누적 집계, `protean.trace.metrics.enabled=true` 일 때만 채워짐) - `modules` — 현재 `ModuleStatus[]` - `summary` — `TraceSummary`: 콘솔 헤더용 **윈도** 집계(아래 참고) - `/platform/traces` 와 마찬가지로 스트림 연결 자체는 trace 기록에서 제외된다(자기-관측 방지). +`ready` 이벤트는 데이터 프레임이 아니라 **연결 확인 응답(ack)** 이다. 트래픽이 없는 플랫폼에서는 *"스트림이 열렸다"* 와 *"서버가 실제로 돌고 있다"* 가 클라이언트 눈에 똑같이 보이기 때문에 존재한다 — 요청이 들어오기 전까지는 아무것도 도착하지 않는다. 연결마다 한 번씩, `EventSource` 가 재연결할 때도 매번, 초기 `trace` 재생보다 앞서 나간다: + +``` +event: ready +data: {"platform":"protean","platformVersion":"0.1.0","tracesEnabled":true, + "metricsEnabled":false,"buffered":5,"tickMs":1000,"capacity":200} +``` + +여기에는 **연결 시점에 한 번 알아야 하고 이후 스트림으로는 알 수 없는** 값만 싣는다 — 주기적으로 바뀌는 값은 `metrics`·`modules`·`summary` 프레임의 몫이다: + +- `platform` — 구현체 식별자. 언제나 `"protean"`. 이 계약을 여러 구현이 공유하므로, 버전 숫자만 받은 콘솔이 그것이 누구의 버전인지 추측하지 않도록 함께 싣는다. **표시·진단용이며 동작을 분기하는 데 쓰지 않는다.** +- `platformVersion` — 이 라이브러리의 버전. jar manifest 의 `Implementation-Version` 에서 읽는다. manifest 가 없는 레이아웃(exploded classes — 테스트·IDE 실행)에서는 **`null`** 이다. 그 상황에서는 버전을 정말로 모르며, 임의의 대체 문자열은 "지금 어느 버전과 이야기하고 있나" 에 대한 지어낸 답이 된다. 필드 자체는 항상 있으므로 클라이언트가 "없음" 과 "모름" 을 구별할 일이 없다. +- `tracesEnabled` / `metricsEnabled` — `protean.trace.enabled` 와 `protean.trace.metrics.enabled`. **"기록이 꺼져 있다"** 와 **"켜져 있는데 아직 아무 일도 없었다"** 를 가른다. 빈 스트림만으로는 이 둘이 구별되지 않는다. +- `buffered` — 이 프레임 **다음에 오는 `trace` 프레임이 싣는 행 수**이며, 링이 보유한 행 수가 아니다. `protean.trace.capacity` 가 재생 상한 200 을 넘으면 둘이 달라진다. 숫자만 알리고 행은 주지 않는 것은 둘 다 안 하느니만 못하다. +- `tickMs` — 푸시 주기(`1000`). 클라이언트의 침묵 감지 워치독이 가정 대신 계약을 기준으로 자기 값을 정할 수 있게 한다. +- `capacity` — **이 연결 시점의** `protean.trace.capacity`. 클라이언트가 "이보다 과거는 서버에 없다" 를 말할 수 있다. LIVE 키이므로 이후 변경은 이미 연결된 클라이언트에 닿지 않으며, 재연결 때 새 값이 다시 나간다. + +이후 버전에서 `ready` 에 필드가 더해질 수 있다 — 클라이언트는 모르는 필드를 무시해야 한다. + ``` event: trace data: {"seq":13,"method":"GET","uri":"/cp/ping","status":200,"latencyMs":2,"moduleId":"cp-mod"} diff --git a/docs/guide/04-rest-api.md b/docs/guide/04-rest-api.md index 1657aca..ae9ccb2 100644 --- a/docs/guide/04-rest-api.md +++ b/docs/guide/04-rest-api.md @@ -393,16 +393,36 @@ Query per-module aggregate metrics (request count, error rate, latency percentil ### GET `/platform/traces/stream` -Live server-sent-events (SSE) push stream — the console uses it in place of 5-second polling. One connection multiplexes four named event types; a freshly opened connection first receives an initial snapshot of all four, then incremental pushes (roughly once per second). +Live server-sent-events (SSE) push stream — the console uses it in place of 5-second polling. A connection is acknowledged with a `ready` frame and then multiplexes four named event types; a freshly opened connection first receives an initial snapshot of all four, then incremental pushes (roughly once per second). - Produces: `text/event-stream` - Named events: + - `ready` — connection acknowledgement, always the first frame (see below) - `trace` — new `RequestTrace` deltas - `metrics` — a `ModuleMetricsSnapshot[]` update (per-module cumulative aggregate; populated only when `protean.trace.metrics.enabled=true`) - `modules` — the current `ModuleStatus[]` - `summary` — a `TraceSummary`: a **windowed** aggregate for the console header (see below) - Like `/platform/traces`, the stream connection itself is excluded from trace recording (no self-observation). +The `ready` event is an acknowledgement, not a data frame. It exists because *"the stream is open"* and *"the server is actually running"* look identical to a client on a quiet platform — nothing else arrives until traffic does. It is sent once per connection, including on every `EventSource` reconnect, ahead of the initial `trace` replay: + +``` +event: ready +data: {"platform":"protean","platformVersion":"0.1.0","tracesEnabled":true, + "metricsEnabled":false,"buffered":5,"tickMs":1000,"capacity":200} +``` + +It carries only what a client must know **once, at connect time** and cannot derive from the stream afterwards — anything that changes on a schedule belongs to the `metrics`/`modules`/`summary` frames instead: + +- `platform` — implementation identifier, always `"protean"`. Present so a console watching several implementations of this contract is not left reading a bare version number and guessing whose it is. **Display and diagnostics only — do not branch behaviour on it.** +- `platformVersion` — this library's version, read from the jar manifest (`Implementation-Version`). **`null`** when running from a layout that has no manifest (exploded classes: tests, IDE runs) — the version is genuinely unknown there, and a placeholder would be a fabricated answer. The field is always present, so a client never has to distinguish "absent" from "unknown". +- `tracesEnabled` / `metricsEnabled` — `protean.trace.enabled` and `protean.trace.metrics.enabled`. They separate **"recording is off"** from **"recording is on but nothing has happened yet"**, which an empty stream cannot. +- `buffered` — how many rows the `trace` frame that follows this one carries, **not** how many the ring holds. The two differ whenever `protean.trace.capacity` exceeds the 200-row replay cap; announcing a count while withholding the rows would be worse than announcing neither. +- `tickMs` — the push period (`1000`), so a client's silence watchdog has a contract to size itself against instead of assuming one. +- `capacity` — `protean.trace.capacity` **as of this connection**, letting a client say "nothing older than this is on the server". It is a live key, so a later change does not reach an already-connected client; the value is re-sent on reconnect. + +Fields may be added to `ready` in later versions — a client must ignore the ones it does not recognize. + ``` event: trace data: {"seq":13,"method":"GET","uri":"/cp/ping","status":200,"latencyMs":2,"moduleId":"cp-mod"} diff --git a/src/main/java/org/htcom/protean/web/StreamReady.java b/src/main/java/org/htcom/protean/web/StreamReady.java new file mode 100644 index 0000000..a87fa8a --- /dev/null +++ b/src/main/java/org/htcom/protean/web/StreamReady.java @@ -0,0 +1,63 @@ +/* + * Copyright (c) 2026 htjulia + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. + */ + +package org.htcom.protean.web; + +/** + * Connection acknowledgement: the first frame of every {@code /platform/traces/stream} connection + * ({@code event: ready}). It exists because "the stream is open" and "the server is actually running" + * are indistinguishable to a client on a quiet platform — nothing else arrives until traffic does. + * + *

This is an ack, not a data frame. It carries only values a client must know once, at connect + * time and cannot derive from the stream afterwards; anything that changes on a schedule belongs to + * the {@code metrics}/{@code modules}/{@code summary} frames instead. + * + * @param platform implementation identifier, constant {@code "protean"}. Present so a console that + * watches several implementations does not have to read a bare version number and + * guess whose it is. Display/diagnostics only — clients must not branch behaviour on it. + * @param platformVersion this library's version, read from the jar manifest ({@code Implementation-Version}). + * {@code null} when running from a layout that has no manifest (exploded classes: tests, + * IDE runs) — the version is genuinely unknown there, and a placeholder string would be + * a fabricated answer to "which version am I talking to". + * @param tracesEnabled {@code protean.trace.enabled}: distinguishes "recording is off" from "recording is on + * but nothing has happened yet", which an empty stream cannot. + * @param metricsEnabled {@code protean.trace.metrics.enabled}: same distinction for the {@code metrics} frame. + * @param buffered how many rows the {@code trace} frame that follows this one carries — not how + * many the ring holds. The two differ whenever the ring is larger than the replay cap, and + * announcing a count while withholding the rows is worse than announcing neither. + * @param tickMs the push period, so a client's silence watchdog has a contract to size itself against + * instead of assuming one. + * @param capacity {@code protean.trace.capacity} as of this connection, letting a client say + * "nothing older than this is on the server". It is a live key, so a later change does not + * reach an already-connected client; the value is re-sent on reconnect. + */ +public record StreamReady( + String platform, + String platformVersion, + boolean tracesEnabled, + boolean metricsEnabled, + int buffered, + long tickMs, + int capacity +) { + + private static final String PLATFORM = "protean"; + + /** Resolved once: a jar manifest cannot change while the JVM runs. */ + private static final String VERSION = version(); + + private static String version() { + Package pkg = StreamReady.class.getPackage(); + return pkg == null ? null : pkg.getImplementationVersion(); + } + + /** Ack for a connection that is about to replay {@code buffered} trace rows. */ + public static StreamReady of(boolean tracesEnabled, boolean metricsEnabled, int buffered, long tickMs, int capacity) { + return new StreamReady(PLATFORM, VERSION, tracesEnabled, metricsEnabled, buffered, tickMs, capacity); + } +} diff --git a/src/main/java/org/htcom/protean/web/TraceStreamService.java b/src/main/java/org/htcom/protean/web/TraceStreamService.java index 45fd1b3..a63fa53 100644 --- a/src/main/java/org/htcom/protean/web/TraceStreamService.java +++ b/src/main/java/org/htcom/protean/web/TraceStreamService.java @@ -44,10 +44,11 @@ * the console; other polling consumers are their own choice). * *

The recording hot path ({@code RequestTraceFilter.record()}) is untouched: this reads the store - * out-of-band on a timer. One SSE connection multiplexes three named events — {@code trace} (incremental, - * only when there is something new), {@code metrics} and {@code modules} (full snapshots each tick, which - * also keeps the connection warm through proxies). A fresh connection receives an initial snapshot of all - * three immediately so it is never blank until the first tick. + * out-of-band on a timer. One SSE connection multiplexes four named events — {@code trace} (incremental, + * only when there is something new), {@code metrics}, {@code modules} and {@code summary} (full snapshots + * each tick, which also keeps the connection warm through proxies). A fresh connection is acknowledged with + * a {@link StreamReady} {@code ready} frame and then receives an initial snapshot of all four immediately, + * so it is never blank until the first tick. * *

Same gating as the admin controllers ({@code @Profile("!worker")} + {@code protean.admin.enabled}), so * it is absent on workers and when the management surface is opted out. @@ -90,7 +91,9 @@ public TraceStreamService(TraceStore store, TraceMetrics metrics, ModulePlatform /** * Opens a live SSE stream. Timeout is disabled (0) because the stream is meant to stay open; if a proxy - * drops it anyway, the browser {@code EventSource} reconnects and gets a fresh initial snapshot. + * drops it anyway, the browser {@code EventSource} reconnects and gets a fresh {@code ready} ack plus a + * fresh initial snapshot — the ack repeats per connection precisely so a reconnecting client re-learns + * the connect-time facts rather than carrying stale ones across the gap. */ public SseEmitter open() { SseEmitter emitter = new SseEmitter(0L); @@ -101,8 +104,11 @@ public SseEmitter open() { }); emitter.onError(ex -> emitters.remove(emitter)); try { + // Read the backlog before the ack so `buffered` can report the rows actually replayed below. + List initial = store.recent(INITIAL_TRACES, null); List mods = moduleStatuses(); - send(emitter, "trace", store.recent(INITIAL_TRACES, null)); + send(emitter, "ready", ready(initial.size())); + send(emitter, "trace", initial); send(emitter, "metrics", metrics.snapshots()); send(emitter, "modules", mods); send(emitter, "summary", summary(mods)); @@ -114,6 +120,17 @@ public SseEmitter open() { return emitter; } + /** + * Connection ack for a client about to be replayed {@code buffered} rows. Every value is read here, at + * connect time, because all four config keys are live — a value cached at startup would be a stale answer + * to a question the client only gets to ask once. + */ + private StreamReady ready(int buffered) { + ProteanProperties.Trace trace = props.getTrace(); + return StreamReady.of(trace.isEnabled(), trace.getMetrics().isEnabled(), buffered, TICK_MS, + trace.getCapacity()); + } + /** One push cycle: advance the cursor by the newest delta, then fan the deltas + fresh snapshots out. */ private void tick() { try { diff --git a/src/test/java/org/htcom/protean/TraceStreamE2ETest.java b/src/test/java/org/htcom/protean/TraceStreamE2ETest.java index d686ba1..27574ec 100644 --- a/src/test/java/org/htcom/protean/TraceStreamE2ETest.java +++ b/src/test/java/org/htcom/protean/TraceStreamE2ETest.java @@ -8,6 +8,8 @@ package org.htcom.protean; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.web.server.LocalServerPort; @@ -26,6 +28,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -34,8 +37,8 @@ /** * Real-JVM e2e for the console SSE push transport ({@code GET /platform/traces/stream}). Verifies the - * multiplexed initial snapshot (trace/metrics/modules named events) and that the server-side ticker keeps - * pushing without any client polling. + * connection ack ({@code ready}) and the multiplexed initial snapshot that follows it, the ack's payload + * contract, and that the server-side ticker keeps pushing without any client polling. */ @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) class TraceStreamE2ETest { @@ -51,36 +54,64 @@ static void props(DynamicPropertyRegistry r) { int port; private final HttpClient client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(5)).build(); + private static final ObjectMapper MAPPER = new ObjectMapper(); @Test - void stream_emits_initial_snapshot_and_keeps_ticking() throws Exception { - // Collect enough events to cover the synchronous initial snapshot (trace+metrics+modules = 3) plus at - // least one later ticker cycle, proving push continues with no client polling. - List events = readSseEventNames(6, Duration.ofSeconds(5)); - - // Initial snapshot delivers all three multiplexed event types. - assertThat(events).contains("trace", "metrics", "modules"); - // The first three are exactly the initial snapshot, in order. - assertThat(events.subList(0, 3)).containsExactly("trace", "metrics", "modules"); + void stream_acks_then_emits_initial_snapshot_and_keeps_ticking() throws Exception { + // Collect enough frames to cover the synchronous ack + initial snapshot (ready+trace+metrics+modules + // +summary = 5) plus at least one later ticker cycle, proving push continues with no client polling. + List events = readSse(8, Duration.ofSeconds(5)).stream().map(Map.Entry::getKey).toList(); + + // The ack comes first, then the four snapshot frames, in this exact order — the console reads the ack + // to tell "the server is live" from "the socket merely opened", so anything ahead of it defeats that. + assertThat(events.subList(0, 5)).containsExactly("ready", "trace", "metrics", "modules", "summary"); // The ticker pushed at least one more cycle after the initial snapshot (no polling involved). - assertThat(events.size()).isGreaterThan(3); + assertThat(events.size()).isGreaterThan(5); + // The ack is per connection, not per tick. + assertThat(events.stream().filter("ready"::equals)).hasSize(1); + } + + @Test + void ready_ack_carries_the_connect_time_contract() throws Exception { + List> frames = readSse(2, Duration.ofSeconds(5)); + + JsonNode ready = MAPPER.readTree(frames.get(0).getValue()); + assertEquals("protean", ready.path("platform").asText()); + // Present but null when running from exploded classes (no jar manifest to read the version from); the + // field must still exist so a client never has to distinguish "absent" from "unknown". + assertThat(ready.has("platformVersion")).isTrue(); + assertThat(ready.path("tracesEnabled").asBoolean()).isTrue(); + assertThat(ready.path("metricsEnabled").asBoolean()).isFalse(); // protean.trace.metrics.enabled default + assertEquals(1_000L, ready.path("tickMs").asLong()); // the watchdog contract + assertEquals(200, ready.path("capacity").asInt()); // protean.trace.capacity default + + // buffered counts the rows of the trace frame that follows, not the rows held by the ring. Announcing a + // count the client does not then receive is the failure this assertion exists to catch. + JsonNode traces = MAPPER.readTree(frames.get(1).getValue()); + assertEquals(traces.size(), ready.path("buffered").asInt()); } - /** Opens the SSE stream and returns the {@code event:} names in arrival order, up to {@code expected} or the deadline. */ - private List readSseEventNames(int expected, Duration deadline) throws Exception { + /** + * Opens the SSE stream and returns {@code (event, data)} pairs in arrival order, up to {@code expected} or + * the deadline. + */ + private List> readSse(int expected, Duration deadline) throws Exception { HttpRequest req = HttpRequest.newBuilder(URI.create("http://localhost:" + port + "/platform/traces/stream")) .header("Accept", "text/event-stream").GET().build(); HttpResponse resp = client.send(req, HttpResponse.BodyHandlers.ofInputStream()); assertEquals(200, resp.statusCode()); assertThat(resp.headers().firstValue("Content-Type").orElse("")).contains("text/event-stream"); - List names = new ArrayList<>(); + List> frames = new ArrayList<>(); CompletableFuture reader = CompletableFuture.runAsync(() -> { try (BufferedReader br = new BufferedReader(new InputStreamReader(resp.body(), StandardCharsets.UTF_8))) { - String line; - while (names.size() < expected && (line = br.readLine()) != null) { + String line, name = null; + while (frames.size() < expected && (line = br.readLine()) != null) { if (line.startsWith("event:")) { - names.add(line.substring("event:".length()).trim()); + name = line.substring("event:".length()).trim(); + } else if (line.startsWith("data:") && name != null) { + frames.add(Map.entry(name, line.substring("data:".length()).trim())); + name = null; } } } catch (Exception ignored) { @@ -92,6 +123,6 @@ private List readSseEventNames(int expected, Duration deadline) throws E reader.cancel(true); } resp.body().close(); - return names; + return frames; } }