fix(stream): emit SSE heartbeats so idle streams survive proxy timeouts - #346
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds stream keepalive configuration, a bucketed heartbeat wheel, StreamHandler support for periodic SSE comments, startup wiring to run the heartbeater, and documentation updates covering the new ChangesSSE keepalive heartbeat
Sequence Diagram(s)sequenceDiagram
participant Client
participant StreamHandler
participant Heartbeater
participant Bucket
participant Subscriber
participant ResponseWriter
Client->>StreamHandler: GET /v1/stream
StreamHandler->>ResponseWriter: write : connected
StreamHandler->>Heartbeater: Add(subscriber)
loop idle period
Heartbeater->>Bucket: Push(: heartbeat)
Bucket->>Subscriber: Send(: heartbeat)
Subscriber-->>StreamHandler: frame bytes from Frames()
StreamHandler->>ResponseWriter: write frame
StreamHandler->>ResponseWriter: Flush()
end
Client-->>StreamHandler: cancel context
StreamHandler->>Heartbeater: Remove(subscriber)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
✨ Simplify code
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. Comment |
|
📚 Docs preview is live → https://3896ebe2-wavehouse-docs.wave-rf.workers.dev
|
Code Coverage OverviewLanguages: Go GoThe overall coverage in the Show a code coverage summary of the most impacted files.
Updated |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/api/stream.go (1)
167-171:⚠️ Potential issue | 🟠 Major | ⚡ Quick winEnd
SSE.PushEventspans on filtered events.On Line [170], the
continuepath bypassespushSpan.End(), so filtered events leak spans on long-lived streams.🩹 Proposed fix
_, pushSpan := tracer.Start(parentCtx, "SSE.PushEvent") out := h.applyStreamPolicy(envelope.Payload, role, claims) if out == nil { + pushSpan.End() continue } id := extractEventTimestamp(out) _, _ = fmt.Fprintf(w, "id: %s\ndata: %s\n\n", id, out) flusher.Flush() // A real delivery already kept the connection warm; restart the // idle clock so heartbeats fire only on genuinely quiet streams. heartbeat.Reset(interval) pushSpan.End()Also applies to: 179-179
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 4d0229d3-0016-4bd7-b1ca-5591243fe593
📒 Files selected for processing (6)
CHANGELOG.mddocs/src/content/docs/api.mddocs/src/content/docs/deployment.mddocs/src/content/docs/reverse-proxy.mdxinternal/api/stream.gointernal/api/stream_test.go
📜 Review details
⏰ Context from checks skipped due to timeout of 300000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (5)
- GitHub Check: Coverage
- GitHub Check: Docs build
- GitHub Check: E2E tests
- GitHub Check: Integration tests
- GitHub Check: Lint
🧰 Additional context used
📓 Path-based instructions (4)
docs/src/content/docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (AGENTS.md)
docs/src/content/docs/**/*.{md,mdx}: Author Mermaid diagrams vertically (flowchart TB/TD) to fit page column width (~46–58rem); reserve LR for genuinely short chains (≤3–4 nodes)
Keep Mermaid node labels short; use
for a second line rather than one long line; lean on semantic node classes (wh, win, pain, fail, infra, neutral, store, client)
Never sit two large diagrams side-by-side; wrap comparisons in…to stack them vertically
Files:
docs/src/content/docs/deployment.mddocs/src/content/docs/api.mddocs/src/content/docs/reverse-proxy.mdx
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: Use Go 1.26 with strict formatting enforced by gofumpt
Use structured logging with log/slog (JSON handler)
Use Chi v5 for HTTP routing
Return errors, don't panic. Wrap with fmt.Errorf("context: %w", err)
Use package naming: lowercase, single word (or abbreviated). internal/ enforces module privacy
No global state: Dependencies are passed explicitly (constructor injection)
Comment the why, not the what. Add a comment only when the reason isn't obvious from the code; a line that matches the surrounding pattern needs none. Keep comments to 1–2 lines
DRY — one source of truth. Before adding logic, look for an existing helper, type, or constant to reuse; before duplicating a rule, factor it into one place every caller reads
Leave it neater than you found it — within reason. Fix small, safe things in passing: a stale comment, an obvious typo, a misnamed local, dead code on your path
Files:
internal/api/stream_test.gointernal/api/stream.go
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Chi HTTP router, JWT/JWKS middleware (from auth/), ingest/query/structured-query/SSE/schema/DLQ/policy/pipes handlers, Hub
Files:
internal/api/stream_test.gointernal/api/stream.go
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Use table-driven tests with tests := []struct{ name string; ... } and t.Run(tt.name, ...)
Use shared mocks from internal/testutil/ (MockPublisher, MockCache, MockDeduplicator, MockSubscriber) instead of creating ad-hoc mocks
Use testutil.MakeJWT(t, claims) and testutil.MakeExpiredJWT(t, claims) for auth tests
Use testutil.NewTestSchemaRegistry(tables) or discovery.NewSchemaRegistryFromMap(tables) for schema-aware tests
Use policy.NewMemoryStore(p) for in-memory policy testing without NATS
Use pipes.NewMemoryStore(queries...) for in-memory pipes testing without NATS
Use testutil.AssertJSONResponse(t, rec, status, expected) and testutil.AssertJSONContains(t, rec, status, substring) for response assertions
Files:
internal/api/stream_test.go
🧠 Learnings (3)
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
docs/src/content/docs/deployment.mddocs/src/content/docs/api.mdCHANGELOG.md
📚 Learning: 2026-05-20T01:02:00.784Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 164
File: internal/api/router_test.go:289-350
Timestamp: 2026-05-20T01:02:00.784Z
Learning: In WaveHouse’s internal API tests (files matching internal/api/**/*_test.go), follow the existing separation-of-concerns convention for testing the RequireRole middleware: inject `ContextKeyRole` directly into the request `context.Context` instead of using `testutil.MakeJWT`/JWT-driven flows. Do not refactor role-gate tests to use JWT tokens—JWT parsing and token handling are covered separately in `middleware_test.go` (the dedicated JWT parsing tests), and mixing those concerns would expand the failure surface and reduce isolation.
Applied to files:
internal/api/stream_test.go
📚 Learning: 2026-05-23T01:23:59.268Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 174
File: internal/api/ingest_test.go:111-111
Timestamp: 2026-05-23T01:23:59.268Z
Learning: In WaveHouse Go tests in internal/api/**/*_test.go, use internal/testutil.AssertJSONErrorResponse(t, w) for HTTP error-path JSON assertions. Do not use (or reintroduce) package-local assertJSONErrorResponse helpers. AssertJSONErrorResponse verifies the response Content-Type is application/json, includes the X-Content-Type-Options: nosniff header, and that the JSON body contains an "error" field.
Applied to files:
internal/api/stream_test.go
🔇 Additional comments (5)
internal/api/stream_test.go (1)
9-10: LGTM!Also applies to: 208-234
CHANGELOG.md (1)
15-16: LGTM!Also applies to: 43-44
docs/src/content/docs/api.md (1)
519-519: LGTM!Also applies to: 535-536
docs/src/content/docs/deployment.md (1)
332-332: LGTM!docs/src/content/docs/reverse-proxy.mdx (1)
59-59: LGTM!Also applies to: 69-79, 136-138, 199-199
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/api/stream.go (1)
132-167:⚠️ Potential issue | 🟠 Major | ⚡ Quick winHeartbeats are not idle-gated, so active streams still flush keepalives.
The select loop writes heartbeat bytes whenever
c.hbChfires, but real event delivery (Line 144+) never re-arms heartbeat scheduling. So even high-traffic streams still emit periodic heartbeat flushes, adding avoidable write pressure.💡 Proposed fix (re-arm after real traffic)
case data := <-ch: var envelope struct { TraceHeaders map[string]string `json:"trace_headers"` Payload []byte `json:"payload"` } @@ id := extractEventTimestamp(out) _, _ = fmt.Fprintf(w, "id: %s\ndata: %s\n\n", id, out) flusher.Flush() + if h.Heartbeater != nil { + // Re-arm heartbeat after actual traffic so keepalives remain idle-only. + h.Heartbeater.Remove(c) + h.Heartbeater.Add(c) + } pushSpan.End()internal/api/stream_test.go (1)
208-238: 🛠️ Refactor suggestion | 🟠 MajorConvert to table-driven test with idle and active-stream scenarios; avoid fixed sleep timing.
This test violates the coding guideline for
**/*_test.gofiles, which require table-driven tests withtests := []struct{ name string; ... }andt.Run(tt.name, ...)— a pattern already used elsewhere ininternal/api/(e.g.,cache_key_test.go). Additionally, it depends on fixedtime.Sleep(200*time.Millisecond)which is fragile under CI load, and covers only the idle scenario; the active-stream behavior (heartbeats should NOT emit on active traffic) is not validated, risking regression.Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 7ff00115-5b75-4453-aa3d-29ee7d115aaf
📒 Files selected for processing (11)
CHANGELOG.mdcmd/wavehouse/main.godocs/src/content/docs/api.mddocs/src/content/docs/configuration.mdxdocs/src/content/docs/reverse-proxy.mdxinternal/api/heartbeat.gointernal/api/heartbeat_test.gointernal/api/stream.gointernal/api/stream_test.gointernal/config/config.gointernal/config/config_test.go
📜 Review details
⏰ Context from checks skipped due to timeout of 300000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (5)
- GitHub Check: Coverage
- GitHub Check: Docs build
- GitHub Check: E2E tests
- GitHub Check: Lint
- GitHub Check: Analyze (actions)
🧰 Additional context used
📓 Path-based instructions (5)
**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*.go: Use Go 1.26 with strict formatting enforced by gofumpt
Use structured logging with log/slog (JSON handler)
Use Chi v5 for HTTP routing
Return errors, don't panic. Wrap with fmt.Errorf("context: %w", err)
Use package naming: lowercase, single word (or abbreviated). internal/ enforces module privacy
No global state: Dependencies are passed explicitly (constructor injection)
Comment the why, not the what. Add a comment only when the reason isn't obvious from the code; a line that matches the surrounding pattern needs none. Keep comments to 1–2 lines
DRY — one source of truth. Before adding logic, look for an existing helper, type, or constant to reuse; before duplicating a rule, factor it into one place every caller reads
Leave it neater than you found it — within reason. Fix small, safe things in passing: a stale comment, an obvious typo, a misnamed local, dead code on your path
Files:
internal/config/config_test.gocmd/wavehouse/main.gointernal/api/stream_test.gointernal/config/config.gointernal/api/heartbeat_test.gointernal/api/heartbeat.gointernal/api/stream.go
internal/config/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
YAML + env var config loading (cleanenv); add field with yaml, env, and env-default tags
Files:
internal/config/config_test.gointernal/config/config.go
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
**/*_test.go: Use table-driven tests with tests := []struct{ name string; ... } and t.Run(tt.name, ...)
Use shared mocks from internal/testutil/ (MockPublisher, MockCache, MockDeduplicator, MockSubscriber) instead of creating ad-hoc mocks
Use testutil.MakeJWT(t, claims) and testutil.MakeExpiredJWT(t, claims) for auth tests
Use testutil.NewTestSchemaRegistry(tables) or discovery.NewSchemaRegistryFromMap(tables) for schema-aware tests
Use policy.NewMemoryStore(p) for in-memory policy testing without NATS
Use pipes.NewMemoryStore(queries...) for in-memory pipes testing without NATS
Use testutil.AssertJSONResponse(t, rec, status, expected) and testutil.AssertJSONContains(t, rec, status, substring) for response assertions
Files:
internal/config/config_test.gointernal/api/stream_test.gointernal/api/heartbeat_test.go
docs/src/content/docs/**/*.{md,mdx}
📄 CodeRabbit inference engine (AGENTS.md)
docs/src/content/docs/**/*.{md,mdx}: Author Mermaid diagrams vertically (flowchart TB/TD) to fit page column width (~46–58rem); reserve LR for genuinely short chains (≤3–4 nodes)
Keep Mermaid node labels short; use
for a second line rather than one long line; lean on semantic node classes (wh, win, pain, fail, infra, neutral, store, client)
Never sit two large diagrams side-by-side; wrap comparisons in…to stack them vertically
Files:
docs/src/content/docs/reverse-proxy.mdxdocs/src/content/docs/api.mddocs/src/content/docs/configuration.mdx
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Chi HTTP router, JWT/JWKS middleware (from auth/), ingest/query/structured-query/SSE/schema/DLQ/policy/pipes handlers, Hub
Files:
internal/api/stream_test.gointernal/api/heartbeat_test.gointernal/api/heartbeat.gointernal/api/stream.go
🧠 Learnings (3)
📚 Learning: 2026-05-20T01:02:00.784Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 164
File: internal/api/router_test.go:289-350
Timestamp: 2026-05-20T01:02:00.784Z
Learning: In WaveHouse’s internal API tests (files matching internal/api/**/*_test.go), follow the existing separation-of-concerns convention for testing the RequireRole middleware: inject `ContextKeyRole` directly into the request `context.Context` instead of using `testutil.MakeJWT`/JWT-driven flows. Do not refactor role-gate tests to use JWT tokens—JWT parsing and token handling are covered separately in `middleware_test.go` (the dedicated JWT parsing tests), and mixing those concerns would expand the failure surface and reduce isolation.
Applied to files:
internal/api/stream_test.gointernal/api/heartbeat_test.go
📚 Learning: 2026-05-23T01:23:59.268Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 174
File: internal/api/ingest_test.go:111-111
Timestamp: 2026-05-23T01:23:59.268Z
Learning: In WaveHouse Go tests in internal/api/**/*_test.go, use internal/testutil.AssertJSONErrorResponse(t, w) for HTTP error-path JSON assertions. Do not use (or reintroduce) package-local assertJSONErrorResponse helpers. AssertJSONErrorResponse verifies the response Content-Type is application/json, includes the X-Content-Type-Options: nosniff header, and that the JSON body contains an "error" field.
Applied to files:
internal/api/stream_test.gointernal/api/heartbeat_test.go
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
docs/src/content/docs/api.mdCHANGELOG.md
🔇 Additional comments (7)
internal/config/config.go (1)
33-33: LGTM!Also applies to: 108-111, 202-207
internal/config/config_test.go (1)
107-140: LGTM!cmd/wavehouse/main.go (1)
362-365: LGTM!CHANGELOG.md (1)
43-43: LGTM!docs/src/content/docs/api.md (1)
519-519: LGTM!docs/src/content/docs/configuration.mdx (1)
42-48: LGTM!Also applies to: 223-226, 297-299
docs/src/content/docs/reverse-proxy.mdx (1)
71-71: LGTM!Also applies to: 137-138, 199-199
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/config/config_test.go (1)
117-139: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winConvert these subtests to a table-driven test.
The assertions are the same harness repeated three times with different inputs, which is exactly the pattern the repo asks to keep table-driven in
*_test.gofiles.Suggested refactor
func TestValidate_KeepaliveValues(t *testing.T) { t.Parallel() base := func() Config { return Config{ Server: Server{Port: 8080}, ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: 30 * time.Second}, Schema: Schema{RefreshInterval: 60}, } } - t.Run("negative interval is rejected", func(t *testing.T) { - t.Parallel() - cfg := base() - cfg.Stream.KeepaliveInterval = -time.Second - err := cfg.Validate() - require.Error(t, err) - assert.Contains(t, err.Error(), "keepalive_interval") - }) - - t.Run("negative buckets is rejected", func(t *testing.T) { - t.Parallel() - cfg := base() - cfg.Stream.KeepaliveBuckets = -1 - err := cfg.Validate() - require.Error(t, err) - assert.Contains(t, err.Error(), "keepalive_buckets") - }) - - t.Run("zero means use default, not an error", func(t *testing.T) { - t.Parallel() - cfg := base() // Stream left at zero values - assert.NoError(t, cfg.Validate()) - }) + tests := []struct { + name string + mutate func(*Config) + wantErr string + }{ + { + name: "negative interval is rejected", + mutate: func(cfg *Config) { + cfg.Stream.KeepaliveInterval = -time.Second + }, + wantErr: "keepalive_interval", + }, + { + name: "negative buckets is rejected", + mutate: func(cfg *Config) { + cfg.Stream.KeepaliveBuckets = -1 + }, + wantErr: "keepalive_buckets", + }, + { + name: "zero means use default, not an error", + mutate: func(cfg *Config) {}, + wantErr: "", + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + cfg := base() + tt.mutate(&cfg) + + err := cfg.Validate() + if tt.wantErr == "" { + assert.NoError(t, err) + return + } + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + }) + } }As per coding guidelines,
**/*_test.go: Use table-driven tests witht.Run(tt.name, ...)and add corresponding tests for each new function.Source: Coding guidelines
internal/api/stream_test.go (1)
210-281: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winUse the repo's table-driven pattern for these new SSE tests.
These scenarios are added as standalone tests, but this path's guideline requires
*_test.gochanges to uset.Run(tt.name, ...). As per coding guidelines,**/*_test.go: "Use table-driven tests witht.Run(tt.name, ...)and add corresponding tests for each new function."Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 453ca959-1209-4d1a-8a84-83774a78bb95
📒 Files selected for processing (20)
AGENTS.mdCHANGELOG.mdcmd/wavehouse/main.goconfig.yamldocs/src/content/docs/api.mddocs/src/content/docs/architecture.mddocs/src/content/docs/configuration.mdxdocs/src/content/docs/deployment.mddocs/src/content/docs/reverse-proxy.mdxinternal/api/stream.gointernal/api/stream_test.gointernal/config/config.gointernal/config/config_test.gointernal/stream/bucket.gointernal/stream/bucket_test.gointernal/stream/doc.gointernal/stream/heartbeat.gointernal/stream/heartbeat_test.gointernal/stream/subscriber.gointernal/stream/subscriber_test.go
📜 Review details
⏰ Context from checks skipped due to timeout. (6)
- GitHub Check: Integration tests
- GitHub Check: E2E tests
- GitHub Check: Docs build
- GitHub Check: Coverage
- GitHub Check: Unit tests
- GitHub Check: Lint
🧰 Additional context used
📓 Path-based instructions (8)
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Use table-driven tests with
t.Run(tt.name, ...)and add corresponding tests for each new function.
Files:
internal/stream/subscriber_test.gointernal/config/config_test.gointernal/stream/bucket_test.gointernal/stream/heartbeat_test.gointernal/api/stream_test.go
**/*.{go,ts,tsx,js,jsx}
📄 CodeRabbit inference engine (AGENTS.md)
Use the repository’s established formatting and naming conventions, including
gofumpt-compatible Go formatting and camelCase where appropriate.
Files:
internal/stream/subscriber_test.gointernal/stream/doc.gointernal/config/config_test.gointernal/stream/subscriber.gointernal/stream/bucket_test.gointernal/config/config.gointernal/stream/bucket.gointernal/stream/heartbeat.gointernal/stream/heartbeat_test.gocmd/wavehouse/main.gointernal/api/stream.gointernal/api/stream_test.go
docs/src/content/docs/deployment.md
📄 CodeRabbit inference engine (AGENTS.md)
docs/src/content/docs/deployment.md: Document deployment-related configuration and schema changes in the deployment docs.
Update the deployment docs when the ingest/event format or ClickHouse insert columns change.
Files:
docs/src/content/docs/deployment.md
AGENTS.md
📄 CodeRabbit inference engine (AGENTS.md)
When changing or adding a core package or architecture invariant, update
AGENTS.mdso the repository instructions stay aligned with the codebase structure.
Files:
AGENTS.md
internal/config/config.go
📄 CodeRabbit inference engine (AGENTS.md)
internal/config/config.go: Keep configuration struct tags (yaml,env, andenv-default) in sync with the documented configuration behavior.
Keep configuration struct tags and the documented configuration surface aligned.
Files:
internal/config/config.go
config.yaml
📄 CodeRabbit inference engine (AGENTS.md)
Keep the example/default configuration file aligned with any config option changes.
Files:
config.yaml
docs/src/content/docs/api.md
📄 CodeRabbit inference engine (AGENTS.md)
Update the API docs for any new or modified endpoint, response shape, or ingest/event format.
Files:
docs/src/content/docs/api.md
docs/src/content/docs/architecture.md
📄 CodeRabbit inference engine (AGENTS.md)
When changing or adding a core package or architecture invariant, update the architecture documentation to keep the named invariant in sync.
Files:
docs/src/content/docs/architecture.md
🧠 Learnings (4)
📓 Common learnings
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: Validate locally before every push by running `make ci` the documented way; do not rely on CI as the first feedback loop.
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: On PR branches, run the full pre-push reviewer flow (`/prepush`) and ensure every required reviewer in `scripts/pre-push-reviewers.sh` reaches `ship_it` or is deliberately skipped on the record before pushing.
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: Every code change must include its corresponding documentation updates and a `CHANGELOG.md` update in the same PR.
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: Address every review finding substantively; either fix it, track it in an issue, mention the bot, and resolve the thread.
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: Agents must create draft PRs with a Conventional Commits-compliant title no longer than 72 characters, and validate the title with `scripts/lint-pr-title.sh` before creating the PR.
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: Never force-push or rebase a PR branch; merge `origin/main` instead when syncing with upstream.
Learnt from: CR
Repo: Wave-RF/WaveHouse
Timestamp: 2026-06-25T22:09:02.439Z
Learning: Do not hand-write markers or use `--no-verify`; use the documented hooks and skip commands instead.
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
docs/src/content/docs/deployment.mdAGENTS.mdCHANGELOG.mddocs/src/content/docs/api.mddocs/src/content/docs/architecture.md
📚 Learning: 2026-05-20T01:02:00.784Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 164
File: internal/api/router_test.go:289-350
Timestamp: 2026-05-20T01:02:00.784Z
Learning: In WaveHouse’s internal API tests (files matching internal/api/**/*_test.go), follow the existing separation-of-concerns convention for testing the RequireRole middleware: inject `ContextKeyRole` directly into the request `context.Context` instead of using `testutil.MakeJWT`/JWT-driven flows. Do not refactor role-gate tests to use JWT tokens—JWT parsing and token handling are covered separately in `middleware_test.go` (the dedicated JWT parsing tests), and mixing those concerns would expand the failure surface and reduce isolation.
Applied to files:
internal/api/stream_test.go
📚 Learning: 2026-05-23T01:23:59.268Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 174
File: internal/api/ingest_test.go:111-111
Timestamp: 2026-05-23T01:23:59.268Z
Learning: In WaveHouse Go tests in internal/api/**/*_test.go, use internal/testutil.AssertJSONErrorResponse(t, w) for HTTP error-path JSON assertions. Do not use (or reintroduce) package-local assertJSONErrorResponse helpers. AssertJSONErrorResponse verifies the response Content-Type is application/json, includes the X-Content-Type-Options: nosniff header, and that the JSON body contains an "error" field.
Applied to files:
internal/api/stream_test.go
🪛 LanguageTool
docs/src/content/docs/reverse-proxy.mdx
[style] ~105-~105: A comma is missing here.
Context: ...le configs ship ~50s) | raise both, e.g. timeout server 1h (note: `timeout tun...
(EG_NO_COMMA)
🔇 Additional comments (5)
internal/stream/subscriber.go (1)
10-45: LGTM!internal/stream/bucket.go (1)
10-62: LGTM!internal/stream/heartbeat_test.go (1)
100-105: 📐 Maintainability & Code QualityNo action needed The repo targets Go 1.26.4, so both
for range Nandt.Context()are supported here.internal/stream/bucket_test.go (1)
71-75: 📐 Maintainability & Code QualityNo change needed — the module targets Go 1.26.4 and the builder image uses Go 1.26, so these integer-range loops are supported.
internal/api/stream_test.go (1)
214-215: 🎯 Functional CorrectnessNo Go toolchain change needed go.mod pins Go 1.26.4, and CI/container builds follow that version, so
t.Context()andfor range connsare supported.
Addresses CodeRabbit review feedback on PR #346, plus a verification pass (subagents re-fetched every cited vendor doc) on the "Idle timeouts by provider" tables — the values and links were sloppy. Citation corrections (reverse-proxy.mdx): - Cloudflare proxied 100s → ~120s (current Proxy Read Timeout). - Cloudflare Tunnel: the cited `keepAliveTimeout` / `--proxy-keepalive-timeout` tunes idle origin-pool reuse, NOT the edge idle reset that drops a quiet stream — reworded to the ~120s edge reset (Enterprise-only), same mechanism as a 524. - Azure Front Door 60s → 30s (the 60s default isn't in any official doc; the docs state 30s). - AWS API Gateway "29s hard" → "29s default max; raisable for Regional/private since 2024, hard only for edge-optimized". - Traefik "none" → no response-write timeout (`writeTimeout` 0); the old claim was wrong and its link was a dead redirect. - Re-pointed links that resolved but didn't document the value (ingress-nginx → ConfigMap, Apache → core `Timeout`, ALB/NLB/CloudFront → the pages that state the default, GCP → docs.cloud.google.com host, Railway → the SSE guide, Fastly → first-byte timeout) and added the `tcp_keepalive_time` man-page citation. - Dropped two rows we can't honestly cite: Akamai (cited page states 5s, not 120s) and Mobile/CGNAT (RFC 6888 doesn't state the 35–65s figure, and that figure is a UDP measurement — SSE is TCP). - Softened the "clears CDN cases" overstatement; added Azure Front Door to the sub-30s list. Code: - NewHeartbeater: clamp `buckets` to 1 when `period < buckets`, so the effective period stays ≈ period in the sub-nanosecond-per-tick edge case instead of ballooning to period × buckets (+test). - config_test `TestValidate_KeepaliveValues` → table-driven. - architecture.md: add `stream/` to the package tree (a second enumeration still omitted it). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XwhWrnP4q4634V3bjm1ZjJ
Follow-up on the keepalive PR (#226), from review discussion on the new stream package. Observability: - Add internal/stream/metrics.go: active streams (UpDownCounter), stream lifetime (histogram), and frames/bytes sent (counters, kind=keepalive vs kind=event). Nil-safe, so the handler holds one unconditionally and tests skip wiring it. Constructed in main.go after InitProvider; recorded at the handler's connect/disconnect and write sites. Tracing cleanup: - Drop the per-event SSE.PushEvent span. The router already excludes /v1/stream from HTTP tracing; a span per delivered event per subscriber is high-volume, low-value, and existed only to read the hub's trace_headers envelope. With it gone the envelope is pure overhead (base64-wraps payload and double-marshals every broadcast), so Hub.Broadcast now sends the raw event bytes and the handler's live case is symmetric with the replay path. Hygiene: - Tighten the stream package + handler comments to terse godoc (one-line summaries; "why" only where non-obvious — the ring clamp, the snapshot-then-send fan-out). #294 context now lives once in doc.go. - main.go: group the heartbeater construct/field/launch lines so the goroutine launch isn't wedged between handler field assignments. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XwhWrnP4q4634V3bjm1ZjJ
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/api/stream.go (1)
124-129: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winRegister the keepalive subscriber before the replay path.
Lines 124-129 run only after the synchronous
replayFromNATSblock above. On reconnects with a slow or empty replay, the handler can still sit silent long enough for an intermediary to idle-close the stream before it ever joins the wheel. Move theNewSubscriber/Add/defer Removeblock above gap fill so heartbeats cover that window too.Suggested fix
h.Metrics.ConnOpened() connectedAt := time.Now() defer func() { h.Metrics.ConnClosed(time.Since(connectedAt)) }() + + sub := stream.NewSubscriber() + if h.Heartbeater != nil { + h.Heartbeater.Add(sub) + defer h.Heartbeater.Remove(sub) + } // Subscribe for live events. ch := make(chan []byte, 64) h.Hub.Subscribe(topic, ch) defer h.Hub.Unsubscribe(topic, ch) @@ - // Register with the shared keepalive wheel so a quiet stream isn't idle-closed - // by a proxy/tunnel between events. - sub := stream.NewSubscriber() - if h.Heartbeater != nil { - h.Heartbeater.Add(sub) - defer h.Heartbeater.Remove(sub) - } - for {
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: b2c949a3-bbc6-40d4-927f-866dcc156bb9
📒 Files selected for processing (17)
AGENTS.mdCHANGELOG.mdcmd/wavehouse/main.godocs/src/content/docs/architecture.mddocs/src/content/docs/configuration.mdxdocs/src/content/docs/reverse-proxy.mdxinternal/api/hub.gointernal/api/hub_test.gointernal/api/stream.gointernal/config/config_test.gointernal/stream/bucket.gointernal/stream/doc.gointernal/stream/heartbeat.gointernal/stream/heartbeat_test.gointernal/stream/metrics.gointernal/stream/metrics_test.gointernal/stream/subscriber.go
📜 Review details
⏰ Context from checks skipped due to timeout. (4)
- GitHub Check: E2E tests
- GitHub Check: Coverage
- GitHub Check: Docs build
- GitHub Check: Lint
⚠️ CI failures not shown inline (2)
GitHub Actions: PR housekeeping / PR housekeeping: fix(stream): emit SSE heartbeats so idle streams survive proxy timeouts
Conclusion: failure
##[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 / 0_PR housekeeping.txt: fix(stream): emit SSE heartbeats so idle streams survive proxy timeouts
Conclusion: failure
##[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 (9)
**/*
📄 CodeRabbit inference engine (AGENTS.md)
**/*: Before pushing a PR branch, run/prepush(which discovers the required reviewers fromscripts/pre-push-reviewers.sh) and satisfy every reviewer that applies to the change.
Every code change must update its corresponding documentation andCHANGELOG.mdin the same PR.
Address and resolve every review finding with a substantive reply, a fix, or a tracked issue; never silently drop review comments.
Create PRs as drafts only, and ensure the PR title passes the Conventional Commits gate and stays within 72 characters.
Never force-push or rebase a PR branch; mergeorigin/maininstead when syncing with upstream changes.
Never hand-write marker files or use--no-verify; use the documented gates and helper scripts instead.
Files:
internal/stream/doc.goAGENTS.mdinternal/stream/metrics.gointernal/stream/bucket.gocmd/wavehouse/main.gointernal/config/config_test.gointernal/stream/metrics_test.gointernal/stream/subscriber.godocs/src/content/docs/configuration.mdxdocs/src/content/docs/reverse-proxy.mdxdocs/src/content/docs/architecture.mdinternal/api/hub.gointernal/stream/heartbeat_test.goCHANGELOG.mdinternal/stream/heartbeat.gointernal/api/stream.gointernal/api/hub_test.go
internal/stream/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
SSE fan-out primitives must keep the subscriber queue, bucket fan-out, heartbeater, and stream metrics behavior intact.
Files:
internal/stream/doc.gointernal/stream/metrics.gointernal/stream/bucket.gointernal/stream/metrics_test.gointernal/stream/subscriber.gointernal/stream/heartbeat_test.gointernal/stream/heartbeat.go
AGENTS.md
📄 CodeRabbit inference engine (AGENTS.md)
Keep
CLAUDE.mdas a thin pointer and avoid duplicating the guidance elsewhere.
Files:
AGENTS.md
cmd/wavehouse/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
cmd/wavehouseshould remain a thin binary entry point and only contain wiring/bootstrap logic.
Files:
cmd/wavehouse/main.go
internal/config/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Configuration structs must use the defined YAML/env/env-default tags and stay in sync with documented config options.
Files:
internal/config/config_test.go
**/*_test.go
📄 CodeRabbit inference engine (AGENTS.md)
Write tests in table-driven form using
t.Run(tt.name, ...), and add test coverage for every new function.
Files:
internal/config/config_test.gointernal/stream/metrics_test.gointernal/stream/heartbeat_test.gointernal/api/hub_test.go
docs/src/content/docs/**/*.md
📄 CodeRabbit inference engine (AGENTS.md)
When authoring Mermaid diagrams, prefer top-down layouts, keep labels short, and avoid side-by-side large diagrams.
Files:
docs/src/content/docs/architecture.md
docs/src/content/docs/architecture.md
📄 CodeRabbit inference engine (AGENTS.md)
When changing core packages or architecture invariants, update the architecture documentation to preserve the named invariant index and rationale.
Files:
docs/src/content/docs/architecture.md
internal/api/**/*.go
📄 CodeRabbit inference engine (AGENTS.md)
Use Chi v5 routing and keep HTTP handlers, middleware, and route registration in the
internal/apilayer.
Files:
internal/api/hub.gointernal/api/stream.gointernal/api/hub_test.go
🧠 Learnings (4)
📚 Learning: 2026-06-10T15:01:09.027Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 312
File: docs/src/content/docs/development.md:0-0
Timestamp: 2026-06-10T15:01:09.027Z
Learning: In this repo’s Markdown review (all .md files), do not flag capitalization/style issues for literal paths starting with ".github/" (or any substring that is a path beginning with ".github/"). Treat ".github" as the correct lowercase dotfile directory name, even when it appears inside prose or code spans; automated checks such as LanguageTool’s "(GITHUB)" rule commonly produce false positives for this literal filesystem path.
Applied to files:
AGENTS.mddocs/src/content/docs/architecture.mdCHANGELOG.md
📚 Learning: 2026-06-26T12:23:22.696Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 346
File: internal/stream/subscriber_test.go:9-28
Timestamp: 2026-06-26T12:23:22.696Z
Learning: In this Go repository, prefer table-driven tests (e.g., `[]struct{...}` with `t.Run(...)`) only for tests that cover multiple scenarios/inputs and can be cleanly enumerated. Do not artificially rewrite a clear single-scenario sequential behavioral-flow test into a table-driven form just to fit the pattern; if there’s only one meaningful scenario, keep the test as a straightforward linear flow (as in `TestSubscriber_SendDeliversThenDropsWhenFull`).
Applied to files:
internal/config/config_test.gointernal/stream/metrics_test.gointernal/stream/heartbeat_test.gointernal/api/hub_test.go
📚 Learning: 2026-05-20T01:02:00.784Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 164
File: internal/api/router_test.go:289-350
Timestamp: 2026-05-20T01:02:00.784Z
Learning: In WaveHouse’s internal API tests (files matching internal/api/**/*_test.go), follow the existing separation-of-concerns convention for testing the RequireRole middleware: inject `ContextKeyRole` directly into the request `context.Context` instead of using `testutil.MakeJWT`/JWT-driven flows. Do not refactor role-gate tests to use JWT tokens—JWT parsing and token handling are covered separately in `middleware_test.go` (the dedicated JWT parsing tests), and mixing those concerns would expand the failure surface and reduce isolation.
Applied to files:
internal/api/hub_test.go
📚 Learning: 2026-05-23T01:23:59.268Z
Learnt from: EricAndrechek
Repo: Wave-RF/WaveHouse PR: 174
File: internal/api/ingest_test.go:111-111
Timestamp: 2026-05-23T01:23:59.268Z
Learning: In WaveHouse Go tests in internal/api/**/*_test.go, use internal/testutil.AssertJSONErrorResponse(t, w) for HTTP error-path JSON assertions. Do not use (or reintroduce) package-local assertJSONErrorResponse helpers. AssertJSONErrorResponse verifies the response Content-Type is application/json, includes the X-Content-Type-Options: nosniff header, and that the JSON body contains an "error" field.
Applied to files:
internal/api/hub_test.go
🪛 LanguageTool
docs/src/content/docs/reverse-proxy.mdx
[style] ~105-~105: A comma is missing here.
Context: ...ng); examples ship ~50s | set both, e.g. timeout server 1h (timeout tunnel i...
(EG_NO_COMMA)
🔇 Additional comments (8)
internal/config/config_test.go (1)
117-151: LGTM!docs/src/content/docs/configuration.mdx (1)
49-49: LGTM!docs/src/content/docs/reverse-proxy.mdx (1)
71-75: LGTM!Also applies to: 94-94, 103-118, 120-121, 131-135, 247-247
docs/src/content/docs/architecture.md (1)
64-65: LGTM!Also applies to: 79-94
AGENTS.md (1)
29-44: LGTM!CHANGELOG.md (1)
43-43: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winMatch the changelog to the actual keepalive payload.
The implementation sends a bare
:SSE comment frame, not: heartbeat, so this entry currently describes the wrong wire output.Suggested fix
-- **SSE streams emit a periodic keepalive comment so quiet connections survive proxy and tunnel idle timeouts** (`internal/stream/` (new package: `subscriber.go`, `bucket.go`, `heartbeat.go` + tests), `internal/api/{stream,stream_test}.go`, `internal/config/{config,config_test}.go`, `cmd/wavehouse/main.go`, `config.yaml`, `docs/src/content/docs/{reverse-proxy.mdx,api.md,configuration.mdx,architecture.md,deployment.md}`): closes `#226`. The stream handler (`internal/stream/metrics.go`, `internal/api/{hub,hub_test}.go` also touched) wrote a single `: connected` comment on open and then sent nothing until an event arrived, so on a quiet table an intermediary's idle timeout reset the connection — Cloudflare's edge (and a Cloudflare Tunnel) dropped quiet streams about every two minutes in dogfooding, and every `curl`/server-side reconnect re-ran NATS gap-fill (browser `EventSource` masked it by auto-reconnecting). A single shared `Heartbeater` goroutine now drives keepalives for every live connection: connections are spread across a ring of buckets and one bucket is pushed a minimal `:` SSE keepalive comment per tick — the writes don't all fire at the same instant, and the per-connection period comes from one timer instead of a `time.Ticker` per connection. +- **SSE streams emit a periodic keepalive comment so quiet connections survive proxy and tunnel idle timeouts** (`internal/stream/` (new package: `subscriber.go`, `bucket.go`, `heartbeat.go` + tests), `internal/api/{stream,stream_test}.go`, `internal/config/{config,config_test}.go`, `cmd/wavehouse/main.go`, `config.yaml`, `docs/src/content/docs/{reverse-proxy.mdx,api.md,configuration.mdx,architecture.md,deployment.md}`): closes `#226`. The stream handler (`internal/stream/metrics.go`, `internal/api/{hub,hub_test}.go` also touched) wrote a single `: connected` comment on open and then sent nothing until an event arrived, so on a quiet table an intermediary's idle timeout reset the connection — Cloudflare's edge (and a Cloudflare Tunnel) dropped quiet streams about every two minutes in dogfooding, and every `curl`/server-side reconnect re-ran NATS gap-fill (browser `EventSource` masked it by auto-reconnecting). A single shared `Heartbeater` goroutine now drives keepalives for every live connection: connections are spread across a ring of buckets and one bucket is pushed a minimal `:` SSE keepalive comment per tick — the writes don't all fire at the same instant, and the per-connection period comes from one timer instead of a `time.Ticker` per connection.> Likely an incorrect or invalid review comment.internal/api/hub.go (1)
57-67: LGTM!internal/stream/metrics.go (1)
12-70: LGTM!
Review follow-up (CodeRabbit + Eric) on the SSE metrics commit. - stream.go: the live-event write now returns on error like the keepalive write, so a vanished client ends the handler (and stops counting bytes that never reached the socket) instead of lingering until another signal. - Centralize the #294 forward-reference in doc.go; the per-symbol comments in subscriber.go / bucket.go / heartbeat.go and the two handler cases now describe only today's behavior, and the remaining over-long comments are trimmed. - hub_test.go: assert the exact raw payload bytes, so the raw-broadcast contract can't silently regress to a re-wrapped envelope. - metrics_test.go: assert the collected datapoints — active back to 0, the duration histogram count/sum, and per-kind frame/byte counts — not just the instrument names. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XwhWrnP4q4634V3bjm1ZjJ
|
Follow-up commits on the keepalive work — Two rounds landed on top of the keepalive wheel:
@coderabbitai — re your outside-diff note on registering the keepalive subscriber before the replay path: declined, with reasoning. Keepalives are only written inside the select loop, which runs after the synchronous gap-fill, so moving registration earlier writes no keepalive during replay (it would at most buffer one frame in the cap-1 queue). And there's no real idle window to cover: an empty gap-fill returns within the ~500ms CI is green and all review threads are resolved. |
|
Tip For best results, initiate chat on the files or code changes.
You’re right that in The metrics/tracing cleanup and the live-path write-error symmetry also look good. I’m happy to withdraw that earlier outside-diff concern. ✏️ Learnings added
|
Addresses @taitelee's review on stream.go: the gap-fill/replay writes weren't recording FrameSent, so the frame/byte totals silently undercounted whatever a reconnect replays. - Add a kind=replay label and record each replayed frame, so reconnect catch-up volume is counted but stays separable from live-tail delivery. - Extract the two identical replay callbacks into one sendReplay closure, which also stops the gap-fill on a write error (client gone) instead of writing into a dead socket — the same asymmetry the keepalive/event cases already handle. - Doc-sync: architecture.md + CHANGELOG now list all three kinds. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XwhWrnP4q4634V3bjm1ZjJ
486324d
Summary
GET /v1/streamwrote a single: connectedcomment on open and then nothing until an event arrived, so on a quiet table an intermediary's idle timeout reset the connection — Cloudflare's edge (and a Cloudflare Tunnel) dropped quiet streams about every two minutes during internal usage, and eachcurl/server-side reconnect re-ran NATS gap-fill (browserEventSourcemasked it via auto-reconnect).This adds a periodic minimal
:SSE keepalive comment to keep idle streams alive. Instead of a timer per connection, a single sharedHeartbeatergoroutine — started once on the server's context inmain.go— drives a ring of buckets: each stream registers into the bucket that fires last (a full rotation before its first ping) and deregisters on disconnect, while every tick pushes the comment to one bucket and advances the hand. The user-facing knob is the effective per-connection period,stream.keepalive_interval(default 30s), with the per-tick interval derived askeepalive_interval ÷ keepalive_buckets, so one rotation always spans the interval; 30s clears the common 55–60s nginx/ingress-nginx/ALB/Heroku idle windows with ~2× margin. The push is a non-blocking send to each subscriber's buffered outbound queue; the owning handler goroutine performs the actual write and flush, so one slow or dead client never blocks the shared goroutine, and a failed write doubles as a liveness probe that ends the handler once the connection is gone.The fan-out machinery lives in a new
internal/streampackage, one abstraction per file:Subscriber(a per-connection outbound frame queue withSend/Frames), theBucketfan-out primitive (subscriberSet), and theHeartbeaterwheel. The handler's keepalive case is now a payload-agnostic byte-pump, so the delivery-path throughput work (#294) can route projected event frames through the same queue and reuseBucketfor project-once-per-(role, table) delivery.Config:
stream.keepalive_interval/WH_STREAM_KEEPALIVE_INTERVAL(default 30s) andstream.keepalive_buckets/WH_STREAM_KEEPALIVE_BUCKETS(default 3);0falls back to the default and a negative value is rejected at startup. The keepalive is a minimal:comment, ignored byEventSourceand spec-compliant parsers (curljust prints them), so it's transparent to clients and needs no SDK change. Also ends theSSE.PushEventspan on policy-filtered events, which previously returned early without closing it. The reverse-proxy guide gains a per-provider idle-timeout reference table (with how-to-change notes and the absolute-cap exceptions a keepalive can't fix).Related Issues
Closes #226.