-
Notifications
You must be signed in to change notification settings - Fork 1
feat: adr block stream api #53
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
10 commits
Select commit
Hold shift + click to select a range
9eaa45c
fix: posting adrs from the lagecy code
hpsing 6e368af
Merge branch 'main' of github.com:getoptimum/optimum-gateway into fix…
hpsing f5ba11f
Merge branch 'main' of github.com:getoptimum/optimum-gateway into fea…
hpsing da43c96
feat: adr for block stream service
hpsing de8a70c
feat: adr for block stream service
hpsing 4271f8f
feat: adr for block stream service
hpsing fce713d
Potential fix for pull request finding
hpsing def8676
fix: suggested changes
hpsing 3ce2e4e
Merge branch 'feat/adr-block-stream-api' of github.com:getoptimum/opt…
hpsing 2c7162e
fix: suggested changes
hpsing File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,183 @@ | ||
| # ADR-0011: Gateway consumer block-stream API (WebSocket + gRPC) | ||
|
|
||
| **Status:** Approved (implementation pending) | ||
| **Date:** 2026-08-05 | ||
|
|
||
| ## Context | ||
|
|
||
| The gateway decodes each beacon-block arrival in `processBeaconBlockArrival()` | ||
| (`pkg/service/gossipsub-gateway/beacon_block_measures.go`), fed by the | ||
| `clMessages` and `mumP2PMessages` channels. We want operators to expose that | ||
| already-decoded stream to their own downstream consumers over WebSocket or gRPC, | ||
| gated per consumer. | ||
|
|
||
| Today the only HTTP surface is `pkg/routes/base.go` (`/health`, `/metrics`, | ||
| `/api/v1/self_info`, `GETOnly`) — there is no inbound-consumer path. | ||
|
|
||
| Two constraints shape the design: | ||
|
|
||
| * **New trust axis.** Existing auth (`auth_token`, `jwks_verifier`) is the | ||
| gateway ↔ control-plane relationship: `OPT_API_KEY` mints the gateway's own | ||
| JWT, and peer JWTs are verified at the handshake (`aud=p2p` / `services`). | ||
| Consumers are a different relationship — operator ↔ its own consumers — and | ||
| must never see `OPT_API_KEY`. They get their own credential. | ||
| * **Never backpressure the mesh.** Relay latency is the SLA, so fan-out to | ||
| consumers must be non-blocking against the ingest goroutines. A stalled client | ||
| must not slow forwarding. | ||
|
|
||
| ## Decision | ||
|
|
||
| Add an opt-in, read-only consumer stream, off by default (same opt-in shape as | ||
| the Obol overlay). Four parts. | ||
|
|
||
| ### 1. Broadcast hub (`pkg/service/streamhub`) | ||
|
|
||
| `processBeaconBlockArrival` runs **once per source observation** — before the | ||
| gateway's cross-path XXHash dedup (`isDuplicateMessage`) — so a block seen via | ||
| both libp2p and mump2p yields two events, one per `source`. The hub emits one | ||
| `BlockEvent` per observation into a fan-out. Event identity is | ||
| `(slot, proposer_index)`; consumers correlate the libp2p and mump2p views of the | ||
| same block by that identity and tell them apart by `source`. | ||
|
|
||
| The stream carries a small transport-neutral frame union, encoded per transport | ||
| (JSON/text over WS, proto over gRPC): | ||
|
|
||
| * `BlockEvent` — a block observation (metadata or raw; see Data model). | ||
| * `lagged` — a control frame sent after ring-buffer overflow, carrying the | ||
| connection's cumulative `dropped` count so the consumer knows it missed events. | ||
|
|
||
| Each subscriber has a bounded ring buffer (default 64). On overflow the hub drops | ||
| the oldest event, increments the per-connection `dropped` counter, and sends a | ||
| `lagged` frame. The emit from ingest is a non-blocking send — it never waits on a | ||
| consumer, so a slow/stalled subscriber cannot backpressure ingest. At-most-once, | ||
| no replay (see Non-goals). | ||
|
|
||
| ### 2. Two transports, one hub | ||
|
|
||
| Both read-only — consumers cannot publish into the mesh. | ||
|
|
||
| * **WebSocket** — `GET /api/v1/stream/blocks` on its own listener | ||
| (`OPT_STREAM_ADDR`) and Fiber app, so `/metrics` and `/health` stay off the | ||
| exposed port. Params: `mode=metadata|raw`, `topics=beacon_block`. | ||
| * **gRPC** — `BlockStream.Subscribe(SubscribeRequest) returns (stream | ||
| BlockEvent)` on `OPT_STREAM_GRPC_ADDR`, from a new | ||
| `proto/getoptimum/optimum_gateway/service/stream/v1/stream.proto`. | ||
|
|
||
| ### 3. Auth — reuse the JWKS verifier | ||
|
|
||
| Consumers present a JWT minted by `auth.getoptimum.io` for a new audience | ||
| `stream`, verified against the JWKS the gateway already caches | ||
| (`OPT_REMOTE_AUTH_URL`) via `pkg/service/jwks_verifier`. Add | ||
| `AudStream = "stream"` next to `AudP2P` / `AudServices`. | ||
|
|
||
| * Token in the `Authorization` header for gRPC metadata and non-browser WS. | ||
| Browsers cannot set WS request headers, so the token rides | ||
| `Sec-WebSocket-Protocol`: the client offers two subprotocol values — a marker | ||
| (`optimum.stream.v1`) and `bearer.<jwt>` — and the server authenticates from | ||
| the `bearer.` value, selects **only the marker** as the negotiated subprotocol, | ||
| and **never** echoes the token back as the selected subprotocol. Auth is | ||
| verified **before** the subscriber is created / the WS upgrade completes; | ||
| unauthenticated connections are rejected, never subscribed. | ||
| * No scope claim in v1 — `aud=stream` is the authorization. There is one topic | ||
| (`beacon_block`), so a valid stream token grants read of the whole stream. The | ||
| gateway still caps connections per `sub` and globally and rate-limits events | ||
| per connection, but those are config-driven, not per-token. (If topics beyond | ||
| `beacon_block` are added later, a scope claim can gate them then.) | ||
| * The verifier sits behind a `ConsumerAuthenticator` interface so an | ||
| operator-local key mode can replace central auth later without touching the | ||
| transport or hub. Interface now, local impl later. | ||
|
|
||
| ### 4. Config (opt-in, off by default) | ||
|
|
||
| | Env / yaml | Default | Purpose | | ||
| | --- | --- | --- | | ||
| | `OPT_STREAM_ENABLE` / `stream_enable` | `false` | Master switch for the consumer API. | | ||
| | `OPT_STREAM_ADDR` / `stream_addr` | `0.0.0.0:9600` | WebSocket/HTTP listener. | | ||
| | `OPT_STREAM_GRPC_ADDR` / `stream_grpc_addr` | `0.0.0.0:9601` | gRPC listener. | | ||
| | `OPT_STREAM_REQUIRE_AUTH` / `stream_require_auth` | `true` | Verify consumer JWTs; `false` only for local dev. | | ||
|
hpsing marked this conversation as resolved.
|
||
| | `OPT_STREAM_MAX_CONNS` / `stream_max_conns` | `256` | Global connection cap. | | ||
| | `OPT_STREAM_MAX_CONNS_PER_SUB` / `stream_max_conns_per_sub` | `8` | Per-subject connection cap. | | ||
| | `OPT_STREAM_BUFFER_SIZE` / `stream_buffer_size` | `64` | Per-connection ring buffer depth (drop-on-overflow). | | ||
|
|
||
| `OPT_REMOTE_AUTH_URL` (already present) supplies the JWKS/issuer. | ||
|
|
||
| **Exposure requirement.** Any non-loopback bind (`stream_addr` / `stream_grpc_addr` | ||
| beyond `127.0.0.1`) requires TLS — native or a trusted TLS-terminating proxy. | ||
| Startup validation must **reject** a non-loopback listener when | ||
| `stream_require_auth=false`; disabling auth is allowed only on a loopback bind for | ||
| local dev. (Read/idle timeouts, max frame size, and the per-connection event-rate | ||
| cap are the other DoS mitigations — see Consequences — with concrete values fixed | ||
| in the implementation.) | ||
|
|
||
| ### Data model — `BlockEvent` | ||
|
|
||
| Two modes, both from the existing decode point: | ||
|
|
||
| * **metadata**: `slot`, `proposer_index`, `parent_root`, `state_root`, | ||
| `block_size_bytes`, `topic`, `source` (`libp2p`|`mump2p`), `received_at_ms`, | ||
| `gateway_id`, `fork_digest`, `stale`. | ||
| * **raw**: the above plus the verbatim `ssz_snappy` bytes. | ||
|
|
||
| `DecodeBeaconBlockHeader` doesn't return a real `body_root`, so `block_root` | ||
| isn't cheaply derivable in metadata mode. It's left to raw mode (consumer-side) | ||
| or a later change rather than adding an SSZ decode on the hot path. | ||
|
|
||
| ## Architecture | ||
|
|
||
| ```mermaid | ||
| flowchart LR | ||
| CL[CL libp2p] --> CH[clMessages] | ||
| MUM[mump2p mesh] --> MCH[mumP2PMessages] | ||
| CH --> DEC[processBeaconBlockArrival<br/>decode once] | ||
| MCH --> DEC | ||
| DEC -->|forward| MESH[relay to mesh / CL] | ||
| DEC -.non-blocking emit.-> HUB[(StreamHub<br/>bounded ring per sub)] | ||
| HUB --> WS[WebSocket server<br/>OPT_STREAM_ADDR] | ||
| HUB --> GRPC[gRPC server<br/>OPT_STREAM_GRPC_ADDR] | ||
| WS --> AUTH{JWKS verify<br/>aud=stream} | ||
| GRPC --> AUTH | ||
| AUTH --> C1[consumer] | ||
| AUTH --> C2[consumer] | ||
| ``` | ||
|
|
||
| ## Alternatives considered | ||
|
|
||
| * **Transport:** WS-only (no typed path) or gRPC-only (not browser-native). | ||
| Chose both on one hub. | ||
| * **Auth:** operator-local static keys or operator-signed JWTs — self-contained, | ||
| but the operator owns key storage, rotation, and revocation. Chose central | ||
| JWKS and kept the `ConsumerAuthenticator` seam for a local mode later. | ||
| * **Payload:** metadata-only (can't reconstruct the block) or raw-only (least | ||
| convenient). Chose both. | ||
|
|
||
| ## Consequences | ||
|
|
||
| * New public surface means DoS exposure. Mitigations: auth-before-subscribe, | ||
| connection caps (global + per-`sub`), per-connection rate cap, read/idle | ||
| timeouts, max frame size, WS keepalive, TLS (proxy or native). | ||
| * Central-auth coupling, bounded by the `ConsumerAuthenticator` seam. | ||
| * Drop-on-lag means slow consumers miss events — surfaced via `lagged`/`dropped` | ||
| rather than silently. | ||
| * Requires the auth service to mint `aud=stream` tokens. | ||
|
|
||
| ## Non-goals (v1) | ||
|
|
||
| * Replay/backfill or a last-N buffer for late joiners. | ||
| * Topics beyond beacon blocks (attestations/aggregated later; a scope claim can | ||
| gate them when they land). | ||
| * Any consumer write path — read-only, always. | ||
|
|
||
| ## Implementation notes | ||
|
|
||
| * Emit `BlockEvent` from `processBeaconBlockArrival` after decode, non-blocking; | ||
| keep `stale` blocks in the stream, flagged, rather than dropping them. | ||
| * New packages `pkg/service/streamhub` and `pkg/service/stream` (WS + gRPC + auth | ||
| middleware), wired in `cmd/main.go` behind `OPT_STREAM_ENABLE`. | ||
| * Extend `pkg/service/jwks_verifier` with `AudStream`, behind | ||
| `ConsumerAuthenticator`. | ||
| * Add `.../stream/v1/stream.proto`; run `make proto`. | ||
| * Tests via `pkg/test_utils` (`jwt_auth_claims.go`, `NewLocalBootstrapServerWithRig`): | ||
| drop-on-lag fan-out, auth-reject-before-upgrade, and non-blocking ingest under | ||
| a stalled consumer. | ||
| * Telemetry: connections (total and per-`sub`), events sent/dropped, auth | ||
| failures, on the existing registry. | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.