perf(stream): project SSE frames once per role, not per subscriber by EricAndrechek · Pull Request #353 · Wave-RF/WaveHouse

EricAndrechek · 2026-06-26T22:12:34Z

What

First PR of the #294 SSE delivery-path throughput epic: move the broadcast hub into internal/stream and project/serialize each live event once per (topic, role) instead of once per subscriber, building on the internal/stream primitives from #346.

Today every connection's read loop independently runs json.Unmarshal → policy.Evaluate → filterEventColumns → json.Marshal (plus a second unmarshal just to read the id: timestamp) on the same event — byte-identical work repeated N times. For a single-role audience (the public dashboard, every viewer public) that's the measured ~2 270 deliveries/s ceiling. This collapses N re-projections to 1.

How

internal/stream/hub.go (new) — Hub. Subscribers register under (topic, role). Broadcast decodes the event once, snapshots the policy once, then per subscribed role applies column policy once, builds one SSE frame, and fans it to that role's Bucket. Skips the decode entirely when nobody is listening.
(topic, role) is the whole key — claims don't matter. The stream path's only per-subscriber transform is column filtering (policy.IsColumnAllowed), which derives solely from the role+table policy entry. Claims feed only the row-level WHERE/CHECK, which the stream path never applies — so the projection is byte-identical for every subscriber of a (role, table). Documented as an invariant in code: if row-level filtering is ever added to streaming, the key must take claims into account.
Unified byte-pump. The handler's two select cases (keepalive vs. per-subscriber event) collapse into one pump over a single Subscriber.Frames() queue of typed Frame{Kind, Data}. The queue grows from cap 1 (keepalive-only) to 64 so live events buffer while the handler is mid-write.
Replay & resumption unchanged. Gap-fill stays per-connection (low-volume, one-time) via the shared stream.ReplayFrame; live frames carry the same id: <received_timestamp>, so Last-Event-ID/?since= resumption and the SSE wire format are byte-for-byte identical (no SDK change).
Slow consumers are now observable. Drops to a full queue increment wavehouse_sse_dropped_frames_total (silent before). An inert Subscriber.Evicted() seam is wired for the eviction follow-up.
Also removes the now-orphaned, test-only internal/api/transform.go (transformForClient had no non-test caller).

Scope / staging

This PR is the architecture shift only. Already done by #346: the per-delivery OTel span (another #294 checkbox). Deferred to follow-ups:

Slow-consumer eviction (feat(observability): latency histograms, error-rate counters, saturation gauges, query-path traces #94 / epic(streaming): SSE delivery-path throughput — project/serialize once per role, not per subscriber #294): consecutive-drop threshold → close Evicted() → handler disconnects → client reconnects + gap-fills.
Buffer right-sizing + lock cost (perf(api): right-size SSE per-subscriber channel buffer (currently hardcoded 64, untested) #152): make the cap-64 queue a config knob.

Testing

internal/stream/{hub,filter,subscriber,bucket,heartbeat}_test.go: project-once-per-role (with a shared-backing-array assertion proving a single serialization), column-filter + table-denial, topic isolation, passthrough/invalid payloads, bucket/topic GC, the drop-metric increment, ReplayFrame, and a concurrent add/remove/broadcast race.
Local make ci green: unit 89.5% / integration / e2e (67 passed) + every coverage gate, all under -race.

Docs

architecture.md (diagram, package tree, stream/ section, Streaming Path), AGENTS.md, CHANGELOG.md, and internal/stream/doc.go are in sync. SSE wire format / event payload (api.md) is unchanged.

Part of #294 (epic — not auto-closed).

🤖 Generated with Claude Code

https://claude.ai/code/session_0131uzPDJtg8As2RnyU5nhUF

Move the broadcast hub into internal/stream as `Hub`: subscribers register under (topic, role), and Broadcast decodes each event once, applies each subscribed role's column policy once, builds one SSE frame per role, and fans it to every member of that role's Bucket. Previously every connection re-ran unmarshal -> Evaluate -> filter -> marshal (plus a second unmarshal for the id: timestamp) on the same event, so the work scaled with subscribers, not distinct output shapes — the ~2270 deliveries/s ceiling from #294. The (topic, role) key is claims-independent: column visibility derives only from the role+table policy entry, and the stream path applies no row-level filter (documented invariant). The handler's two select cases collapse into one byte-pump over a single Subscriber.Frames() queue of typed Frames; the queue grows from cap 1 to 64 so live events buffer while the handler is mid-write. Gap-fill replay and Last-Event-ID/?since= resumption are unchanged (replay stays per-connection via the shared stream.ReplayFrame; live frames carry the same id: <received_timestamp>). Slow-consumer drops now increment wavehouse_sse_dropped_frames_total; an inert Subscriber.Evicted() seam is wired for the eviction follow-up. The per-delivery OTel span (another #294 item) was already removed in #346. Also drops the now-orphaned, test-only internal/api/transform.go (transformForClient had no non-test caller). First PR of the #294 epic. Deferred to follow-ups: active slow-consumer eviction (#94) and right-sizing the subscriber buffer + lock cost (#152). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0131uzPDJtg8As2RnyU5nhUF

coderabbitai · 2026-06-26T22:13:01Z

Warning

Review limit reached

@EricAndrechek, we couldn't start this review because you've reached your PR review rate limit.

More reviews will be available in 24 minutes and 17 seconds. Learn how PR review limits work.

Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file).

⌛ How to resolve this issue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits.

🚦 How do rate limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please see our Fair Usage Limits Policy for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: bdf9de31-dee9-404e-bed2-eed138964403

📥 Commits

Reviewing files that changed from the base of the PR and between 17b88bd and 9ab1a0f.

📒 Files selected for processing (5)

cmd/wavehouse/main.go
internal/api/stream.go
internal/stream/filter_test.go
internal/stream/hub.go
internal/stream/hub_test.go

📝 Walkthrough

Walkthrough

The SSE path now uses internal/stream for role-scoped fan-out, typed frames, keepalive delivery, and dropped-frame metrics. The HTTP handler, MQ bridge, tests, and architecture docs now reference the new hub and filtering names.

Changes

SSE Delivery-Path Refactor

Layer / File(s)	Summary
Typed frames and queueing primitives `internal/stream/subscriber.go`, `internal/stream/bucket.go`, `internal/stream/heartbeat.go`, `internal/stream/metrics.go`, `internal/stream/subscriber_test.go`, `internal/stream/bucket_test.go`, `internal/stream/heartbeat_test.go`	`Subscriber` now carries `Frame` values, `Bucket` fans out snapshots of subscribers, keepalives are emitted as `KindKeepalive` frames, and dropped-frame telemetry is added and tested.
Hub projection and replay `internal/stream/hub.go`, `internal/stream/filter_test.go`	`Hub` now groups subscribers by topic and role, projects raw payloads once per role, applies column filtering and replay projection, and formats SSE wire frames.
Hub behavior tests `internal/stream/hub_test.go`	Hub tests cover per-role fan-out, policy filtering, topic isolation, payload passthrough, cleanup, dropped-frame metrics, replay projection, and concurrent add/broadcast/remove.
API handler and runtime wiring `internal/api/stream.go`, `cmd/wavehouse/main.go`, `internal/api/*_test.go`, `tests/integration/setup_test.go`	The SSE handler, MQ bridge, and integration/router test setup now use `stream.NewHub`, `stream.Subscriber`, shared SSE metrics, and frame-based draining/eviction.
Docs and ownership notes `AGENTS.md`, `CHANGELOG.md`, `docs/src/content/docs/architecture.md`, `internal/policy/policy.go`, `internal/stream/doc.go`	Package ownership notes, architecture text, changelog, and policy comments are updated to use `stream` as the SSE hub home and `filterColumns` as the column-filtering reference.

Sequence Diagram(s)

sequenceDiagram
  participant embeddedMQ as embeddedMQ subscription callback
  participant streamHub as stream.Hub
  participant streamSubscriber as stream.Subscriber
  participant StreamHandler as StreamHandler
  participant Heartbeater as Heartbeater
  embeddedMQ->>streamHub: Broadcast(msg.Subject, msg.Data)
  streamHub->>streamSubscriber: Send(Frame)
  Heartbeater->>streamSubscriber: Send(KindKeepalive frame)
  StreamHandler->>streamSubscriber: read Frames()
  StreamHandler->>streamSubscriber: stop on Evicted()

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

epic(streaming): SSE delivery-path throughput — project/serialize once per role, not per subscriber #294: The PR moves SSE projection and serialization into internal/stream and applies them once per role before fan-out.

Possibly related PRs

Wave-RF/WaveHouse#124: Both PRs change the hub fan-out path; this PR replaces the API hub with internal/stream.Hub, while #124 adjusted internal/api/hub.go fan-out behavior.
Wave-RF/WaveHouse#346: This PR’s typed Frame queues, keepalive framing, and subscriber eviction build on the streaming/heartbeat model introduced there.

Suggested reviewers

taitelee

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 41.67% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the main SSE refactor: projecting frames once per role instead of per subscriber.
Description check	✅ Passed	The description is on-topic and matches the hub refactor, queue changes, metrics, replay behavior, and tests.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

Create PR with unit tests
Commit unit tests in branch sse-fan-out

✨ Simplify code

Create PR with simplified code
Commit simplified code in branch sse-fan-out

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Comment @coderabbitai help to get the list of available commands.}

github-actions · 2026-06-26T22:15:12Z

📚 Docs preview is live → https://554b1433-wavehouse-docs.wave-rf.workers.dev

Commit — 9ab1a0f: fix(stream): address PR perf(stream): project SSE frames once per role, not per subscriber #353 review (fail-closed, replay via Hub, frames)
Author — @EricAndrechek, Claude Opus 4.8 (1M context)
Committed — 2026-06-26 18:39 (UTC-04:00)
Deployed — 2026-06-26 18:51 EDT

coderabbitai

Actionable comments posted: 6

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: d4ab6da2-7dcf-4d42-b5f8-704077280b08

📥 Commits

Reviewing files that changed from the base of the PR and between 725fdee and 17b88bd.

📒 Files selected for processing (25)

AGENTS.md
CHANGELOG.md
cmd/wavehouse/main.go
docs/src/content/docs/architecture.md
internal/api/errors_test.go
internal/api/hub.go
internal/api/hub_test.go
internal/api/router_test.go
internal/api/stream.go
internal/api/stream_test.go
internal/api/transform.go
internal/api/transform_test.go
internal/policy/policy.go
internal/stream/bucket.go
internal/stream/bucket_test.go
internal/stream/doc.go
internal/stream/filter_test.go
internal/stream/heartbeat.go
internal/stream/heartbeat_test.go
internal/stream/hub.go
internal/stream/hub_test.go
internal/stream/metrics.go
internal/stream/subscriber.go
internal/stream/subscriber_test.go
tests/integration/setup_test.go

💤 Files with no reviewable changes (4)

internal/api/transform.go
internal/api/hub.go
internal/api/transform_test.go
internal/api/hub_test.go

📜 Review details

⏰ Context from checks skipped due to timeout. (2)

GitHub Check: Docs preview
GitHub Check: Lint

⚠️ CI failures not shown inline (2)

GitHub Actions: PR housekeeping / 0_PR housekeeping.txt: perf(stream): project SSE frames once per role, not per subscriber

Conclusion: failure

View job details

##[group]Run # Single source of truth for the rule: scripts/lint-pr-title.sh — the
 �[36;1m# Single source of truth for the rule: scripts/lint-pr-title.sh — the�[0m
 �[36;1m# SAME validator the local agent gate runs (.claude/hooks/agent-bash-gate.sh),�[0m
 �[36;1m# so CI and local can't drift. The checkout above is ref: main, so this is�[0m
 �[36;1m# always the default-branch script. Dependabot's grouped-update titles�[0m
 �[36;1m# routinely exceed the 72-char subject cap and the format isn't�[0m
 �[36;1m# configurable, so Dependabot PRs are exempt from the length check�[0m
 �[36;1m# (the format check still applies).�[0m
 �[36;1mif [[ "$PR_AUTHOR" == "dependabot[bot]" || "$PR_AUTHOR" == "app/dependabot" ]]; then�[0m
 �[36;1m  export PR_TITLE_SKIP_LENGTH=1�[0m
 �[36;1mfi�[0m
 �[36;1m�[0m
 �[36;1mif reason=$(bash scripts/lint-pr-title.sh "$PR_TITLE" 2>&1); then�[0m
 �[36;1m  echo "passed=true" >> "$GITHUB_OUTPUT"�[0m
 �[36;1m  echo "PR title OK: $PR_TITLE"�[0m
 �[36;1melse�[0m
 �[36;1m  echo "passed=false" >> "$GITHUB_OUTPUT"�[0m
 �[36;1m  printf '%s\n' "$reason"�[0m
 �[36;1m  echo "::error::$(printf '%s' "$reason" | head -1)"�[0m

GitHub Actions: PR housekeeping / PR housekeeping: perf(stream): project SSE frames once per role, not per subscriber

Conclusion: failure

View job details

##[group]Run # Single source of truth for the rule: scripts/lint-pr-title.sh — the
 �[36;1m# Single source of truth for the rule: scripts/lint-pr-title.sh — the�[0m
 �[36;1m# SAME validator the local agent gate runs (.claude/hooks/agent-bash-gate.sh),�[0m
 �[36;1m# so CI and local can't drift. The checkout above is ref: main, so this is�[0m
 �[36;1m# always the default-branch script. Dependabot's grouped-update titles�[0m
 �[36;1m# routinely exceed the 72-char subject cap and the format isn't�[0m
 �[36;1m# configurable, so Dependabot PRs are exempt from the length check�[0m
 �[36;1m# (the format check still applies).�[0m
 �[36;1mif [[ "$PR_AUTHOR" == "dependabot[bot]" || "$PR_AUTHOR" == "app/dependabot" ]]; then�[0m
 �[36;1m  export PR_TITLE_SKIP_LENGTH=1�[0m
 �[36;1mfi�[0m
 �[36;1m�[0m
 �[36;1mif reason=$(bash scripts/lint-pr-title.sh "$PR_TITLE" 2>&1); then�[0m
 �[36;1m  echo "passed=true" >> "$GITHUB_OUTPUT"�[0m
 �[36;1m  echo "PR title OK: $PR_TITLE"�[0m
 �[36;1melse�[0m
 �[36;1m  echo "passed=false" >> "$GITHUB_OUTPUT"�[0m
 �[36;1m  printf '%s\n' "$reason"�[0m
 �[36;1m  echo "::error::$(printf '%s' "$reason" | head -1)"�[0m

🧰 Additional context used

📓 Path-based instructions (7)

internal/stream/**