From ab41feed6bbbe6370887309f1ed2a1981ba50229 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:35:55 -0400 Subject: [PATCH 01/13] fix(boot): non-fatal schema discovery, /health 503 with diagnostic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, any error from the initial SchemaRegistry.Refresh in cmd/wavehouse — connection-refused, missing database, transient blip — called os.Exit(1). The supervisor would restart the process every ~10s in an unbounded loop, port 8080 never bound, and operators got `connection refused` on probes even though ClickHouse was otherwise healthy. The binary was unrecoverable without intervention. Now the first Refresh failure is non-fatal: a new api.BootState records the diagnostic, the server still binds :8080, and a background goroutine calls SchemaRegistry.RetryRefresh (exp backoff 2s → 60s) until success or shutdown. While BootState.Err() != nil, /health returns 503 with {"status":"degraded","error":"..."} so operators can curl the endpoint instead of grepping a restart-loop log; /ready does the same. Once a Refresh succeeds, BootState.Set(nil) flips both endpoints back to 200 and stays there — transient runtime CH blips after that point are reflected in /ready only. Closes #95. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 3 + cmd/wavehouse/main.go | 41 +++++-- docs/api.md | 16 ++- docs/architecture.md | 4 +- docs/deployment.md | 14 ++- internal/api/health.go | 56 +++++++++ internal/api/health_test.go | 84 ++++++++++++++ internal/discovery/discovery.go | 35 ++++++ internal/discovery/discovery_test.go | 166 +++++++++++++++++++++++++++ 9 files changed, 404 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e434e347..e68dc321 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +### Changed +- **Boot-time schema discovery no longer crashes the binary** (`cmd/wavehouse/main.go`, `internal/api/health.go`, `internal/discovery/discovery.go`, `docs/api.md`, `docs/deployment.md`). Previously, any error from the initial `SchemaRegistry.Refresh` — connection-refused, missing database, transient network blip — caused `cmd/wavehouse` to call `os.Exit(1)`. The supervisor would restart the process every ~10s in an unbounded loop, port 8080 never bound, and operators got `connection refused` on probes even when ClickHouse was otherwise healthy. Now the first Refresh failure is non-fatal: a new `api.BootState` is set with the diagnostic, the server still binds `:8080`, and a background goroutine calls the new `SchemaRegistry.RetryRefresh(ctx, 2s, 60s, onAttempt)` with exponential backoff until success or shutdown. While `BootState.Err()` is non-nil, `/health` returns 503 with `{"status":"degraded","error":"…"}` so an operator can `curl /health` to learn why the gateway isn't accepting traffic instead of grepping a restart-loop log; `/ready` returns 503 with `{"status":"not ready",…}` for the same reason. Once a Refresh succeeds, `BootState.Set(nil)` flips both endpoints back to 200 and stays there for the rest of the process lifetime — transient ClickHouse blips after boot are reflected in `/ready` only. Resolves #95. + ### Removed - **`project-orchestrator.yml` workflow + its three composite-action artifacts** (`.github/workflows/project-orchestrator.yml`, `.github/actions/board-upsert-status/`, `.github/actions/set-linked-issues-status/`, `.github/scripts/board-fetch-item.sh`, `AGENTS.md`, `CHANGELOG.md`): −887 lines net. The orchestrator was the largest single source of cross-trigger complexity on this repo (3-4 workflow_run-chained runs per PR push, `statusCheckRollup` GraphQL perms quirks, integration-token `NONE` for private-org members) for behaviour that is mostly either provided natively by GitHub or a one-click manual operation on a 4-person team. Replaced by: reviewer-assign step in `housekeeping.yml` that fires once on `pull_request_target: opened` / `ready_for_review` (not per-synchronize, so it doesn't re-spam after `dismiss_stale_reviews_on_push`), plus GitHub's native Projects v2 workflows (`Auto-add to project`, `Item added`, `Pull request merged`) configured in the project UI. Trade-offs explicit in the PR body: drafts no longer auto-flip on bot-clean, `CHANGES_REQUESTED` doesn't auto-move the board card, linked-issue card mirroring is dropped. AGENTS.md §"Governance Files" + §"Task Board state machine" + §"Review tooling reference" all rewritten to match. `dependabot-automerge.yml` trimmed in parallel: no more board-upsert step (native handles placement), `PROJECT_BOARD_TOKEN` guard removed (no longer used in this workflow), reviewer list sourced from `board-config.env`'s `ADMINS` via `replace()`, major-bump comment uses the marker-comment upsert pattern from `housekeeping.yml`. - **`STATUS_*` and old `ADMINS` consumers in `board-config.env`** — STATUS option IDs had only orchestrator-side consumers and are now unreferenced. `ADMINS` was restored to `board-config.env` after the initial orchestrator-removal commit dropped it (Gemini and Claude both flagged the resulting drift across three inlined copies); both `housekeeping.yml` and `dependabot-automerge.yml` now load `ADMINS` from `board-config.env`. `admin-approval.yml` keeps its own inline copy with the documented latency-avoidance reasoning. diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 678f0c24..df3cb51b 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -114,12 +114,37 @@ func run() int { } defer func() { _ = chConn.Close() }() - // Schema discovery — discover ClickHouse table schemas on boot. + // Process-lifetime context — cancelled by the SIGINT/SIGTERM handler + // below. Created here (instead of further down) so the boot-time schema + // discovery retry goroutine ties cleanly to shutdown. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Schema discovery — non-fatal on boot. If the first Refresh fails + // (ClickHouse unreachable, database missing, etc.) we mark the binary + // degraded via bootState (which /health surfaces as 503 + diagnostic) + // and retry in the background with exponential backoff. The process + // still binds :8080 so operators can `curl /health` instead of grepping + // a restart-loop log. Once a Refresh succeeds, bootState flips to nil + // and /health returns 200. + bootState := api.NewBootState(nil) refreshInterval := time.Duration(cfg.Schema.RefreshInterval) * time.Second registry := discovery.NewSchemaRegistry(chConn, cfg.ClickHouse.Database, refreshInterval, logger) - if err := registry.Refresh(context.Background()); err != nil { - logger.Error("schema discovery failed on boot", "error", err) - return 1 + if err := registry.Refresh(ctx); err != nil { + logger.Warn("schema discovery failed on boot, retrying in background", "error", err) + bootState.Set(fmt.Errorf("schema discovery: %w", err)) + go func() { + retryErr := registry.RetryRefresh(ctx, 2*time.Second, 60*time.Second, func(attemptErr error) { + logger.Warn("schema discovery retry failed", "error", attemptErr) + bootState.Set(fmt.Errorf("schema discovery: %w", attemptErr)) + }) + if retryErr != nil { + // ctx cancelled before success — process is shutting down. + return + } + logger.Info("schema discovery succeeded after retry, /health now 200") + bootState.Set(nil) + }() } // State directories — one configurable root, fixed subdir convention. @@ -191,9 +216,6 @@ func run() int { // Hub for streaming fan-out. hub := api.NewHub() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - // Start schema auto-refresh. go registry.StartAutoRefresh(ctx) @@ -259,6 +281,9 @@ func run() int { queryHandler := api.NewQueryHandler(chConn, tiered, time.Duration(cfg.Cache.DefaultTTL)*time.Second) queryHandler.PolicyStore = policyStore + healthHandler := api.NewHealthHandler(chConn) + healthHandler.Boot = bootState + sseHandler := api.NewSSEHandler(hub, js) sseHandler.PolicyStore = policyStore @@ -270,7 +295,7 @@ func run() int { Query: queryHandler, SSE: sseHandler, WS: wsHandler, - Health: api.NewHealthHandler(chConn), + Health: healthHandler, Schema: api.NewSchemaHandler(registry), DLQ: dlqHandler, Policy: api.NewPolicyHandler(policyStore), diff --git a/docs/api.md b/docs/api.md index 6bec3636..2317be68 100644 --- a/docs/api.md +++ b/docs/api.md @@ -67,19 +67,29 @@ The per-endpoint error tables below list the bodies you can expect for each stat ### `GET /health` — Liveness Probe -Returns `200 OK` if the process is running. No authentication required. +Returns `200 OK` once the gateway has discovered ClickHouse table schemas at least once. Returns `503 Service Unavailable` with a diagnostic body while the boot-time schema discovery retry loop is still running (ClickHouse unreachable, target database missing, etc.). No authentication required. -**Response:** +**Response (ready):** ```json {"status": "ok"} ``` +**Response (boot-degraded):** + +```json +{"status": "degraded", "error": "schema discovery: dial tcp 127.0.0.1:9000: connect: connection refused"} +``` + +Status code: `503 Service Unavailable` + +The boot-degraded response lets an operator `curl /health` to learn why the gateway hasn't bound traffic yet, instead of grepping a restart-loop log. Schema discovery retries with exponential backoff (2s → 60s); once a Refresh succeeds, `/health` flips to `200` and stays there for the rest of the process lifetime — transient ClickHouse blips after that point are reflected in `/ready`, not `/health`. + --- ### `GET /ready` — Readiness Probe -Returns `200 OK` if the process is running and ClickHouse is reachable. No authentication required. +Returns `200 OK` if the process is fully booted (schema discovery complete) and ClickHouse is currently reachable. Returns `503 Service Unavailable` otherwise. No authentication required. **Response (ready):** diff --git a/docs/architecture.md b/docs/architecture.md index 9d62a1c1..bad51e9e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -95,7 +95,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with standar - **schema.go** — Schema discovery API: list all schemas, get one table, trigger refresh. - **dlq.go** — DLQ stats endpoint and `EnsureDLQStream` helper for creating the `WAVEHOUSE_DLQ` NATS stream. - **hub.go** — In-process pub/sub for broadcasting MQ messages to connected streaming clients. -- **health.go** — Liveness (`/health`) and readiness (`/ready`) probes. +- **health.go** — Liveness (`/health`) and readiness (`/ready`) probes. Both consult an optional `BootState` so they can return 503 with a diagnostic while boot-time schema discovery is still failing in the retry loop (see `cmd/wavehouse/main.go`); once `BootState.Set(nil)` fires, `/health` returns 200 and stays there. ### `cache/` — Two-Tier Caching @@ -114,7 +114,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with standar ### `discovery/` — Schema Discovery & Validation -- **discovery.go** — `SchemaRegistry` queries `system.columns` to discover ClickHouse table schemas. Supports periodic auto-refresh and on-demand refresh. Thread-safe via `sync.RWMutex`. +- **discovery.go** — `SchemaRegistry` queries `system.columns` to discover ClickHouse table schemas. Supports periodic auto-refresh, on-demand refresh, and `RetryRefresh` (boot-time exponential backoff loop used by `cmd/wavehouse` so a transiently unreachable ClickHouse doesn't crash-loop the binary). Thread-safe via `sync.RWMutex`. - **validation.go** — `Validate(schema, data)` checks incoming JSON against the discovered schema: unknown fields, type compatibility, missing required columns, null handling. - **discovery_test.go** — Unit tests for validation logic. diff --git a/docs/deployment.md b/docs/deployment.md index 68a16cb5..3ec9ab14 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -221,11 +221,21 @@ The directory is a *seed*, not authoritative storage: after bootstrap, the API + API servers in standalone mode expose two health endpoints: -- `GET /health` — Liveness probe. Returns 200 if the process is running. No external dependencies. -- `GET /ready` — Readiness probe. Returns 200 if ClickHouse is reachable, 503 otherwise. +- `GET /health` — Liveness probe. Returns 200 once the gateway has discovered ClickHouse table schemas at least once. Returns 503 with a diagnostic body while the boot-time schema discovery retry loop is still running (e.g. ClickHouse unreachable, target database missing). After successful boot, `/health` stays 200 — transient ClickHouse blips at runtime are reflected in `/ready`, not `/health`. +- `GET /ready` — Readiness probe. Returns 200 if the gateway is fully booted and ClickHouse is currently reachable, 503 otherwise. Configure your load balancer or orchestrator to use these endpoints. +### Boot-time degraded mode + +If ClickHouse is unreachable when WaveHouse starts (connection refused, missing database, DNS failure, etc.), the gateway no longer exits — it binds `:8080` and serves `/health` 503 with the latest schema-discovery error as the diagnostic. Schema discovery retries in the background with exponential backoff (2s → 60s cap). Once a Refresh succeeds, `/health` flips to 200 and normal serving begins automatically. + +This means: + +- The supervisor (docker, systemd, k8s) won't restart the binary every ~10s because of a slow-starting or temporarily-unreachable ClickHouse. The process stays up and recovers in place. +- An operator can `curl /health` and read the exact failure mode instead of grepping a restart-loop log. +- `/v1/ingest/{table}` and other schema-aware endpoints will reject requests with a 4xx until discovery succeeds, since the schema registry is empty. + ### Docker `HEALTHCHECK` Both bundled Dockerfiles (`deployments/Dockerfile` and `deployments/Dockerfile.goreleaser`) ship a built-in `HEALTHCHECK` that probes `/health` every 10 seconds. Because the runtime image is distroless (no shell, no `curl`/`wget`), the check uses the binary's own `health` subcommand: diff --git a/internal/api/health.go b/internal/api/health.go index c2c76e8a..84b1fcf5 100644 --- a/internal/api/health.go +++ b/internal/api/health.go @@ -3,13 +3,54 @@ package api import ( "encoding/json" "net/http" + "sync" "github.com/ClickHouse/clickhouse-go/v2/lib/driver" ) +// BootState tracks a one-shot startup diagnostic surfaced by /health. While +// Err() returns non-nil the binary is considered to be in degraded-boot mode: +// /health responds 503 with the diagnostic message instead of 200, so an +// operator can curl the endpoint to learn why the gateway isn't accepting +// traffic yet. Once boot work (today: ClickHouse schema discovery) succeeds, +// Set(nil) flips /health back to 200. +// +// BootState is safe for concurrent use. +type BootState struct { + mu sync.RWMutex + err error +} + +// NewBootState returns a BootState seeded with initialErr. Pass nil if the +// binary is fully ready at construction time; pass a non-nil error to start +// in degraded mode (the goroutine that performs boot work calls Set(nil) on +// success). +func NewBootState(initialErr error) *BootState { + return &BootState{err: initialErr} +} + +// Set replaces the current diagnostic. Pass nil to mark the binary ready. +func (b *BootState) Set(err error) { + b.mu.Lock() + defer b.mu.Unlock() + b.err = err +} + +// Err returns the current diagnostic, or nil if the binary is ready. +func (b *BootState) Err() error { + b.mu.RLock() + defer b.mu.RUnlock() + return b.err +} + // HealthHandler provides liveness and readiness probes. type HealthHandler struct { CHConn driver.Conn + // Boot is consulted by Liveness. When non-nil and its Err() is non-nil, + // Liveness reports 503 with the diagnostic message — used while boot- + // time schema discovery is still failing in the retry loop. A nil Boot + // preserves the pre-retry-loop behaviour (always 200). + Boot *BootState } func NewHealthHandler(chConn driver.Conn) *HealthHandler { @@ -19,6 +60,14 @@ func NewHealthHandler(chConn driver.Conn) *HealthHandler { func (h *HealthHandler) Liveness(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Content-Type-Options", "nosniff") + + if h.Boot != nil { + if err := h.Boot.Err(); err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]string{"status": "degraded", "error": err.Error()}) + return + } + } _ = json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) } @@ -26,6 +75,13 @@ func (h *HealthHandler) Readiness(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Content-Type-Options", "nosniff") + if h.Boot != nil { + if err := h.Boot.Err(); err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]string{"status": "not ready", "error": err.Error()}) + return + } + } if h.CHConn != nil { if err := h.CHConn.Ping(r.Context()); err != nil { w.WriteHeader(http.StatusServiceUnavailable) diff --git a/internal/api/health_test.go b/internal/api/health_test.go index 29eb33e8..7e632ee5 100644 --- a/internal/api/health_test.go +++ b/internal/api/health_test.go @@ -77,3 +77,87 @@ func TestHealth_Readiness_PingFails(t *testing.T) { assert.Equal(t, "not ready", resp["status"]) assert.Equal(t, "ch ping failed", resp["error"]) } + +func TestBootState_DefaultIsReady(t *testing.T) { + t.Parallel() + bs := NewBootState(nil) + assert.NoError(t, bs.Err()) +} + +func TestBootState_SetAndGet(t *testing.T) { + t.Parallel() + bs := NewBootState(errors.New("starting")) + require.Error(t, bs.Err()) + assert.Equal(t, "starting", bs.Err().Error()) + + bs.Set(errors.New("still starting")) + assert.Equal(t, "still starting", bs.Err().Error()) + + bs.Set(nil) + assert.NoError(t, bs.Err()) +} + +func TestHealth_Liveness_BootDegraded(t *testing.T) { + t.Parallel() + // When BootState reports a non-nil error, Liveness must return 503 with + // the diagnostic in the JSON body. This is the behaviour the gateway + // relies on so an operator can `curl /health` during a boot-time + // ClickHouse outage instead of grepping a restart-loop log. + h := NewHealthHandler(nil) + h.Boot = NewBootState(errors.New("schema discovery: dial tcp 127.0.0.1:9000: connect: connection refused")) + + w := httptest.NewRecorder() + r := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/health", nil) + h.Liveness(w, r) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + assert.Equal(t, "nosniff", w.Header().Get("X-Content-Type-Options")) + var resp map[string]string + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + assert.Equal(t, "degraded", resp["status"]) + assert.Contains(t, resp["error"], "connection refused") +} + +func TestHealth_Liveness_BootReadyFlipsTo200(t *testing.T) { + t.Parallel() + // Once the retry loop succeeds, BootState.Set(nil) flips Liveness back + // to 200 — the "normal serving begins" half of the issue's contract. + bs := NewBootState(errors.New("schema discovery: still failing")) + h := NewHealthHandler(nil) + h.Boot = bs + + w := httptest.NewRecorder() + r := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/health", nil) + h.Liveness(w, r) + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + + bs.Set(nil) + + w = httptest.NewRecorder() + r = httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/health", nil) + h.Liveness(w, r) + assert.Equal(t, http.StatusOK, w.Code) + var resp map[string]string + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + assert.Equal(t, "ok", resp["status"]) +} + +func TestHealth_Readiness_BootDegradedReports503(t *testing.T) { + t.Parallel() + // Readiness should also surface boot degradation: a kubelet readiness + // probe needs to see "not ready" while schema discovery is still + // failing, even if the ClickHouse ping path would otherwise succeed. + h := NewHealthHandler(nil) + h.Boot = NewBootState(errors.New("schema discovery: code: 81, Database wavehouse does not exist")) + + w := httptest.NewRecorder() + r := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ready", nil) + h.Readiness(w, r) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + var resp map[string]string + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + assert.Equal(t, "not ready", resp["status"]) + assert.Contains(t, resp["error"], "Database wavehouse does not exist") +} diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index c2a56a63..f45e2a42 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -113,6 +113,41 @@ func (sr *SchemaRegistry) List() []*TableSchema { return result } +// RetryRefresh repeatedly calls Refresh with exponential backoff until it +// succeeds or ctx is cancelled. onAttempt is invoked after each failed +// attempt with the resulting error, letting callers surface the latest +// diagnostic (e.g. via /health) while the registry is still degraded. +// +// The first attempt fires immediately. After a failure the loop sleeps for +// initialBackoff, then doubles up to maxBackoff between attempts. Returns +// nil on success or ctx.Err() on cancellation. Zero/negative bounds are +// clamped to sensible defaults rather than busy-looping. +func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxBackoff time.Duration, onAttempt func(err error)) error { + if initialBackoff <= 0 { + initialBackoff = time.Second + } + if maxBackoff < initialBackoff { + maxBackoff = initialBackoff + } + backoff := initialBackoff + for { + if err := sr.Refresh(ctx); err == nil { + return nil + } else if onAttempt != nil { + onAttempt(err) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoff): + } + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } +} + // StartAutoRefresh runs a background goroutine that refreshes schemas // at the configured interval. Blocks until ctx is cancelled. func (sr *SchemaRegistry) StartAutoRefresh(ctx context.Context) { diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 23dd411f..f6fa9874 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -2,11 +2,14 @@ package discovery import ( "context" + "errors" "io" "log/slog" + "sync/atomic" "testing" "time" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -361,6 +364,169 @@ func TestNewSchemaRegistryFromMap_Empty(t *testing.T) { assert.Nil(t, sr.Get("x")) } +// fakeConn implements just enough of driver.Conn to drive RetryRefresh. It +// returns successive errors from errsThenSuccess, and once that slice is +// exhausted returns emptyRows (which makes Refresh succeed with zero tables). +// The nil-embedded interface trick handles every other method — none of them +// are reached by Refresh. +type fakeConn struct { + driver.Conn + errsThenSuccess []error + calls atomic.Int32 +} + +func (c *fakeConn) Query(_ context.Context, _ string, _ ...any) (driver.Rows, error) { + n := c.calls.Add(1) + if int(n) <= len(c.errsThenSuccess) { + return nil, c.errsThenSuccess[n-1] + } + return &emptyRows{}, nil +} + +type emptyRows struct { + driver.Rows +} + +func (*emptyRows) Next() bool { return false } +func (*emptyRows) Close() error { return nil } +func (*emptyRows) Err() error { return nil } +func (*emptyRows) ColumnTypes() []driver.ColumnType { + return nil +} + +func newFakeRegistry(t *testing.T, errs []error) (*SchemaRegistry, *fakeConn) { + t.Helper() + conn := &fakeConn{errsThenSuccess: errs} + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + return NewSchemaRegistry(conn, "test", time.Hour, logger), conn +} + +// TestRetryRefresh_SucceedsOnFirstAttempt is the happy path: Refresh returns +// nil immediately and no backoff is observed. +func TestRetryRefresh_SucceedsOnFirstAttempt(t *testing.T) { + t.Parallel() + sr, conn := newFakeRegistry(t, nil) + + var attempts int32 + start := time.Now() + err := sr.RetryRefresh(context.Background(), time.Hour, time.Hour, func(error) { + atomic.AddInt32(&attempts, 1) + }) + require.NoError(t, err) + + assert.Less(t, time.Since(start), 100*time.Millisecond, "should not have slept") + assert.Equal(t, int32(0), atomic.LoadInt32(&attempts), "onAttempt should only fire on failure") + assert.Equal(t, int32(1), conn.calls.Load(), "exactly one Query call expected") +} + +// TestRetryRefresh_RetriesUntilSuccess verifies the loop keeps trying through +// transient errors and surfaces each failure via onAttempt with the most +// recent error becoming the diagnostic. +func TestRetryRefresh_RetriesUntilSuccess(t *testing.T) { + t.Parallel() + errFirst := errors.New("dial tcp: connect: connection refused") + errSecond := errors.New("code: 81, Database wavehouse does not exist") + sr, conn := newFakeRegistry(t, []error{errFirst, errSecond}) + + var captured []error + err := sr.RetryRefresh(context.Background(), time.Millisecond, 10*time.Millisecond, func(e error) { + captured = append(captured, e) + }) + require.NoError(t, err) + + assert.Equal(t, int32(3), conn.calls.Load(), "two failures + one success") + require.Len(t, captured, 2) + assert.ErrorIs(t, captured[0], errFirst) + assert.ErrorIs(t, captured[1], errSecond) +} + +// TestRetryRefresh_ReturnsOnContextCancel verifies the loop exits with +// ctx.Err() when the context is cancelled mid-backoff, rather than spinning +// forever or panicking. +func TestRetryRefresh_ReturnsOnContextCancel(t *testing.T) { + t.Parallel() + // Long-running failure stream so success never happens. + errs := make([]error, 1000) + for i := range errs { + errs[i] = errors.New("still failing") + } + sr, _ := newFakeRegistry(t, errs) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + // Long backoff so the loop is parked in the sleep when we cancel. + done <- sr.RetryRefresh(ctx, time.Second, time.Second, nil) + }() + + // Give the goroutine a moment to enter its first sleep. + time.Sleep(50 * time.Millisecond) + cancel() + + select { + case err := <-done: + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled) + case <-time.After(2 * time.Second): + t.Fatal("RetryRefresh did not return after ctx cancel") + } +} + +// TestRetryRefresh_BackoffIsBounded verifies that maxBackoff caps the +// exponential growth. We use small bounds so the test stays fast. +func TestRetryRefresh_BackoffIsBounded(t *testing.T) { + t.Parallel() + // Five failures then success; with initial 1ms and max 4ms backoff, + // sleeps are 1, 2, 4, 4, 4 = 15ms total. We assert it stays well under + // the unbounded-doubling worst case (1+2+4+8+16 = 31ms). + errs := make([]error, 5) + for i := range errs { + errs[i] = errors.New("transient") + } + sr, _ := newFakeRegistry(t, errs) + + start := time.Now() + err := sr.RetryRefresh(context.Background(), time.Millisecond, 4*time.Millisecond, nil) + require.NoError(t, err) + + elapsed := time.Since(start) + // Lower bound proves we actually slept; upper bound proves capping. + assert.GreaterOrEqual(t, elapsed, 10*time.Millisecond) + assert.Less(t, elapsed, 100*time.Millisecond) +} + +// TestRetryRefresh_NilOnAttemptIsSafe verifies the loop tolerates a nil +// onAttempt callback (we want callers to be able to skip the diagnostic +// surface when they don't need it). +func TestRetryRefresh_NilOnAttemptIsSafe(t *testing.T) { + t.Parallel() + sr, _ := newFakeRegistry(t, []error{errors.New("once")}) + + err := sr.RetryRefresh(context.Background(), time.Millisecond, time.Millisecond, nil) + require.NoError(t, err) +} + +// TestRetryRefresh_ClampsInvalidBackoffs verifies zero or negative bounds +// fall back to defaults instead of busy-looping (a 0-duration time.After +// would otherwise fire immediately and spin the CPU). +func TestRetryRefresh_ClampsInvalidBackoffs(t *testing.T) { + t.Parallel() + sr, _ := newFakeRegistry(t, []error{errors.New("once")}) + + // Cancel via context after a short window so we have a hard ceiling on + // how long this test can run, but expect success well before that. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Use 0 / -1 — should clamp to defaults (1s initial), but with only one + // failure the loop succeeds on the second attempt after ~1s. + start := time.Now() + err := sr.RetryRefresh(ctx, 0, -1, nil) + require.NoError(t, err) + elapsed := time.Since(start) + assert.GreaterOrEqual(t, elapsed, 900*time.Millisecond, "should sleep about 1s with clamped default") +} + // TestStartAutoRefresh_ExitsOnContextCancel verifies the ticker loop exits // cleanly when the context is cancelled, without calling Refresh (nil conn // would otherwise panic). From ab6cfbb35d5e713459a8e06c00d3bdf88b18b3b9 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:45:41 -0400 Subject: [PATCH 02/13] test(discovery): drop redundant time.Sleep in cancel test + comment fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Claude review on #125: TestRetryRefresh_ReturnsOnContextCancel used a 50ms sleep before cancel() that violated AGENTS.md ("use channel sync not goroutine scheduling assumptions") and was unnecessary for correctness — `done` is the only sync we need, and the RetryRefresh loop observes ctx.Done() whether cancel fires before or after the goroutine enters its select. Verified with `go test -race -count=10`. Also corrects a stale doc comment on HealthHandler.Boot that said "consulted by Liveness" — Readiness consults it too, by design (a kubelet readiness probe should see "not ready" while boot is failing). Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/api/health.go | 9 +++++---- internal/discovery/discovery_test.go | 6 ++++-- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/internal/api/health.go b/internal/api/health.go index 84b1fcf5..f8264f3b 100644 --- a/internal/api/health.go +++ b/internal/api/health.go @@ -46,10 +46,11 @@ func (b *BootState) Err() error { // HealthHandler provides liveness and readiness probes. type HealthHandler struct { CHConn driver.Conn - // Boot is consulted by Liveness. When non-nil and its Err() is non-nil, - // Liveness reports 503 with the diagnostic message — used while boot- - // time schema discovery is still failing in the retry loop. A nil Boot - // preserves the pre-retry-loop behaviour (always 200). + // Boot is consulted by both Liveness and Readiness. When non-nil and + // its Err() is non-nil, both endpoints report 503 with the diagnostic + // — used while boot-time schema discovery is still failing in the + // retry loop. A nil Boot preserves the pre-retry-loop behaviour + // (Liveness always 200; Readiness 503 only on Ping failure). Boot *BootState } diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index f6fa9874..d270c040 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -459,8 +459,10 @@ func TestRetryRefresh_ReturnsOnContextCancel(t *testing.T) { done <- sr.RetryRefresh(ctx, time.Second, time.Second, nil) }() - // Give the goroutine a moment to enter its first sleep. - time.Sleep(50 * time.Millisecond) + // `done` provides the only synchronisation we need. Whether cancel + // fires before or after the goroutine enters the select, the loop + // must observe ctx.Done() and return — no sleep-based scheduling + // assumption required. cancel() select { From 2c570410d6612e2f59ea3212c399e6df263ddd6a Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:49:00 -0400 Subject: [PATCH 03/13] docs(agents): record non-fatal boot as Key Design Decision #15 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Gemini's review on #125. Boot-time degraded mode + retry loop is a meaningful design decision worth documenting alongside the other 'Async ingestion', 'DLQ', 'Active Sweeper' style entries — future agents working on cmd/wavehouse will want to know /health is allowed to 503 by design during initial Refresh. Co-Authored-By: Claude Opus 4.7 (1M context) --- AGENTS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/AGENTS.md b/AGENTS.md index 1ac1e502..68084259 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,6 +41,7 @@ Ten internal packages under `internal/`: 12. **Structured queries**: Type-safe query AST endpoint (`POST /v1/tables/{table}/query`) validated against schema, with permission enforcement, timestamp bucketing for cache optimization, and `DefaultMaxRows` (10,000) limit cap. 13. **Named query pipes**: Pre-defined SQL templates (inspired by Tinybird) with parameter binding, role restrictions, and caching. Stored in NATS KV with `.sql` file directory bootstrap. 14. **TypeScript SDK**: `@wavehouse/sdk` — zero-dependency client with typed query builder, real-time SSE, live queries with smart aggregation classification (incrementable/decomposable/poll), and codegen CLI. +15. **Non-fatal boot**: Schema-discovery failure on boot (ClickHouse unreachable, missing database, transient network blip) is non-fatal — `cmd/wavehouse` records an `api.BootState` diagnostic, binds `:8080`, and retries via `SchemaRegistry.RetryRefresh` (exp backoff 2s → 60s) in the background. `/health` and `/ready` return 503 with the latest diagnostic until a Refresh succeeds, after which `/health` stays 200 for the rest of the process. Keeps supervisor restart loops bounded and gives operators a queryable failure surface (`curl /health`) instead of a restart-log grep. ## Code Conventions From 2144b3a381b74ddf9750724ef8631a12ec2a8b5e Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Tue, 12 May 2026 18:56:36 -0400 Subject: [PATCH 04/13] fix(docs,test): startupProbe in k8s example + extract clamp helper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two Claude-review fixes from #125: 1) [MUST] docs/deployment.md previously claimed the binary "won't restart" — false for Kubernetes when /health is wired to livenessProbe (the example we ship), since kubelet still kills the pod after failureThreshold × periodSeconds. The retry-loop window can exceed that, recreating the same restart problem the PR is solving. Fixed by (a) softening the prose to be accurate about the binary-vs-orchestrator distinction and (b) adding a startupProbe to the K8s example so liveness and readiness are gated until the first schema discovery succeeds. Docker HEALTHCHECK is unaffected (doesn't restart by default). 2) [SHOULD] TestRetryRefresh_ClampsInvalidBackoffs slept ~1s in the unit suite to observe the clamp-to-default behaviour. Extracted the clamp into a private clampBackoff helper so the invariant is now testable in microseconds via TestClampBackoff (6 table-driven cases). Old slow test removed; discovery coverage rose 89.6% → 89.8%. Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/deployment.md | 12 ++++++-- internal/discovery/discovery.go | 23 +++++++++----- internal/discovery/discovery_test.go | 46 ++++++++++++++++------------ 3 files changed, 53 insertions(+), 28 deletions(-) diff --git a/docs/deployment.md b/docs/deployment.md index 3ec9ab14..8959ada5 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -232,10 +232,12 @@ If ClickHouse is unreachable when WaveHouse starts (connection refused, missing This means: -- The supervisor (docker, systemd, k8s) won't restart the binary every ~10s because of a slow-starting or temporarily-unreachable ClickHouse. The process stays up and recovers in place. +- The binary itself no longer exits and crash-loops every ~10s under a supervisor. Process state is preserved across CH outages. - An operator can `curl /health` and read the exact failure mode instead of grepping a restart-loop log. - `/v1/ingest/{table}` and other schema-aware endpoints will reject requests with a 4xx until discovery succeeds, since the schema registry is empty. +**Important — orchestrator restart semantics.** `/health` returning 503 during the retry window is what most LB / `depends_on` setups want (route around the unready instance, hold dependents), but a Kubernetes `livenessProbe` pointed at `/health` will still mark the pod unhealthy and restart it after `failureThreshold × periodSeconds` elapses (default ~30s) — effectively re-creating the restart loop at a slower cadence. Use a `startupProbe` to gate liveness/readiness until the first successful schema discovery (see the K8s example below). Docker `HEALTHCHECK` marks the container `(unhealthy)` but does not restart it by default, so docker-compose deployments don't need a separate startupProbe-equivalent — the `HEALTHCHECK`'s `--start-period=15s` plus `service_healthy` dependency wait covers the same idea at a smaller scale. + ### Docker `HEALTHCHECK` Both bundled Dockerfiles (`deployments/Dockerfile` and `deployments/Dockerfile.goreleaser`) ship a built-in `HEALTHCHECK` that probes `/health` every 10 seconds. Because the runtime image is distroless (no shell, no `curl`/`wget`), the check uses the binary's own `health` subcommand: @@ -277,15 +279,21 @@ If you need different intervals (e.g. faster probes for E2E tests), override per ### Kubernetes / orchestrator note -K8s `livenessProbe` and `readinessProbe` use kubelet HTTP probes from outside the container — they don't go through the Dockerfile `HEALTHCHECK` at all. Configure them directly against `/health` and `/ready` in the PodSpec: +K8s `livenessProbe` and `readinessProbe` use kubelet HTTP probes from outside the container — they don't go through the Dockerfile `HEALTHCHECK` at all. Configure them directly against `/health` and `/ready` in the PodSpec, and add a `startupProbe` so the boot-time schema-discovery retry window doesn't trip liveness and restart the pod: ```yaml +startupProbe: + httpGet: { path: /health, port: 8080 } + failureThreshold: 30 # allow up to 5 min for first schema discovery (30 × periodSeconds) + periodSeconds: 10 livenessProbe: httpGet: { path: /health, port: 8080 } readinessProbe: httpGet: { path: /ready, port: 8080 } ``` +Until `startupProbe` succeeds, kubelet doesn't run `livenessProbe` or `readinessProbe` against the pod — so a slow or temporarily-unreachable ClickHouse can't restart-loop the pod via the liveness path. Size `failureThreshold` to your expected worst-case CH boot time; the default 30 × 10s = 5min is generous and works for compose-on-NAS-style deployments where CH and WaveHouse can race during a host reboot. + ## ClickHouse Schema WaveHouse uses a **Bring Your Own Schema** model. You create your tables in ClickHouse with whatever columns and engines you need. WaveHouse discovers the schemas automatically via `system.columns` and validates ingest data against them. diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index f45e2a42..e5d7478f 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -113,6 +113,20 @@ func (sr *SchemaRegistry) List() []*TableSchema { return result } +// clampBackoff guards RetryRefresh against busy-looping when callers pass +// zero/negative durations. Any non-positive initialBackoff falls back to 1s, +// and maxBackoff is widened to at least initialBackoff. Extracted so the +// clamp behaviour is unit-testable without observing real wall-clock sleeps. +func clampBackoff(initialBackoff, maxBackoff time.Duration) (time.Duration, time.Duration) { + if initialBackoff <= 0 { + initialBackoff = time.Second + } + if maxBackoff < initialBackoff { + maxBackoff = initialBackoff + } + return initialBackoff, maxBackoff +} + // RetryRefresh repeatedly calls Refresh with exponential backoff until it // succeeds or ctx is cancelled. onAttempt is invoked after each failed // attempt with the resulting error, letting callers surface the latest @@ -121,14 +135,9 @@ func (sr *SchemaRegistry) List() []*TableSchema { // The first attempt fires immediately. After a failure the loop sleeps for // initialBackoff, then doubles up to maxBackoff between attempts. Returns // nil on success or ctx.Err() on cancellation. Zero/negative bounds are -// clamped to sensible defaults rather than busy-looping. +// clamped via clampBackoff rather than busy-looping. func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxBackoff time.Duration, onAttempt func(err error)) error { - if initialBackoff <= 0 { - initialBackoff = time.Second - } - if maxBackoff < initialBackoff { - maxBackoff = initialBackoff - } + initialBackoff, maxBackoff = clampBackoff(initialBackoff, maxBackoff) backoff := initialBackoff for { if err := sr.Refresh(ctx); err == nil { diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index d270c040..85d4684b 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -508,25 +508,33 @@ func TestRetryRefresh_NilOnAttemptIsSafe(t *testing.T) { require.NoError(t, err) } -// TestRetryRefresh_ClampsInvalidBackoffs verifies zero or negative bounds -// fall back to defaults instead of busy-looping (a 0-duration time.After -// would otherwise fire immediately and spin the CPU). -func TestRetryRefresh_ClampsInvalidBackoffs(t *testing.T) { - t.Parallel() - sr, _ := newFakeRegistry(t, []error{errors.New("once")}) - - // Cancel via context after a short window so we have a hard ceiling on - // how long this test can run, but expect success well before that. - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - // Use 0 / -1 — should clamp to defaults (1s initial), but with only one - // failure the loop succeeds on the second attempt after ~1s. - start := time.Now() - err := sr.RetryRefresh(ctx, 0, -1, nil) - require.NoError(t, err) - elapsed := time.Since(start) - assert.GreaterOrEqual(t, elapsed, 900*time.Millisecond, "should sleep about 1s with clamped default") +// TestClampBackoff exercises the busy-loop guard directly without observing +// real wall-clock sleeps. RetryRefresh delegates to this helper, so the +// behavioural guarantee ("zero/negative bounds fall back to a 1s default +// rather than spinning the CPU") is locked in here instead of by sleeping +// 1s in a unit test. +func TestClampBackoff(t *testing.T) { + t.Parallel() + tests := []struct { + name string + initial, maxIn time.Duration + wantI, wantM time.Duration + }{ + {"zero initial clamps to 1s", 0, 5 * time.Second, time.Second, 5 * time.Second}, + {"negative initial clamps to 1s", -1, 5 * time.Second, time.Second, 5 * time.Second}, + {"zero max widens to initial", 100 * time.Millisecond, 0, 100 * time.Millisecond, 100 * time.Millisecond}, + {"max smaller than initial widens to initial", 2 * time.Second, time.Second, 2 * time.Second, 2 * time.Second}, + {"positive bounds pass through", 100 * time.Millisecond, time.Second, 100 * time.Millisecond, time.Second}, + {"both non-positive falls back to default", 0, -1, time.Second, time.Second}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + i, m := clampBackoff(tt.initial, tt.maxIn) + assert.Equal(t, tt.wantI, i) + assert.Equal(t, tt.wantM, m) + }) + } } // TestStartAutoRefresh_ExitsOnContextCancel verifies the ticker loop exits From dcb452bf0bcf1e5294b048f0c915019b91124ecc Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:33:49 -0400 Subject: [PATCH 05/13] fix(boot,docs): address CodeRabbit review on #125 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four CodeRabbit findings on commit 2144b3a: 1) [Minor] CHANGELOG MD022 + sticky-/health vs conditional-/ready wording. The previous prose said both endpoints "stay 200 for the rest of the process lifetime" then contradicted itself by noting /ready reflects transient blips. Rewrote to make explicit: /health is sticky-200 after the first Refresh success (it answers "did boot complete once"); /ready remains conditional on current ClickHouse reachability (it answers "can I serve traffic right now"). Also added the missing blank line above ### Changed and updated the doc paths to the post-Astro-merge locations. 2) [Major] StartAutoRefresh + RetryRefresh race. Two goroutines were calling Refresh concurrently during boot (mutex-safe, but stale bootState if the auto-refresh tick succeeded first). Restructured so StartAutoRefresh only starts after the first successful Refresh (sync or retry). One Refresh caller at a time, no stale-bootState window. 3) [Minor] /health docs bind-state phrasing — "hasn't bound traffic yet" was inaccurate (the binary IS bound on :8080 in boot-degraded mode, it serves /health diagnostics — it just isn't accepting ingest/query traffic). Reworded to "isn't ready to serve traffic yet" with an explicit clarifying sentence. 4) [Major] TestRetryRefresh_BackoffIsBounded wall-clock upper bound tightened to 100ms was too aggressive for shared CI runners — the 15ms-of-real-sleeps budget can stretch past 100ms under scheduler pressure. Relaxed to 250ms, which still catches a real unbounded- backoff regression (orders-of-magnitude blowup) without flaking. Pushed back on two: - testutil.AssertJSONResponse/AssertJSONContains aren't a clean fit for these tests (substring-on-error-message + deliberate header coverage on the 503 branches — see TestHealth_Readiness_PingFails comment). Existing health_test.go tests use the same manual-parsing pattern; one consistent style across the file > selective helper adoption. - 5-into-1 table-driven consolidation of the RetryRefresh tests would produce a struct with mostly-nil optional fields per row (different setup, different assertions, ctx-cancel needs goroutine+chan orchestration). Current form is short, top-down readable. The neighbour TestClampBackoff IS table-driven (6 pure-fn input/output rows) — used where it actually fits. All four Gemini concerns from the earlier review remain resolved (Gemini explicitly acknowledged each pushback in the merged review threads on 2c57041 — sanitization, time.After memory leak, AGENTS.md placement). make verify + make test-unit pass; discovery coverage holds at 89.8%. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 4 +++- cmd/wavehouse/main.go | 10 ++++++---- docs/src/content/docs/api.md | 2 +- internal/discovery/discovery_test.go | 10 +++++++--- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e6f0fbeb..2368d053 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased + ### Changed -- **Boot-time schema discovery no longer crashes the binary** (`cmd/wavehouse/main.go`, `internal/api/health.go`, `internal/discovery/discovery.go`, `docs/api.md`, `docs/deployment.md`). Previously, any error from the initial `SchemaRegistry.Refresh` — connection-refused, missing database, transient network blip — caused `cmd/wavehouse` to call `os.Exit(1)`. The supervisor would restart the process every ~10s in an unbounded loop, port 8080 never bound, and operators got `connection refused` on probes even when ClickHouse was otherwise healthy. Now the first Refresh failure is non-fatal: a new `api.BootState` is set with the diagnostic, the server still binds `:8080`, and a background goroutine calls the new `SchemaRegistry.RetryRefresh(ctx, 2s, 60s, onAttempt)` with exponential backoff until success or shutdown. While `BootState.Err()` is non-nil, `/health` returns 503 with `{"status":"degraded","error":"…"}` so an operator can `curl /health` to learn why the gateway isn't accepting traffic instead of grepping a restart-loop log; `/ready` returns 503 with `{"status":"not ready",…}` for the same reason. Once a Refresh succeeds, `BootState.Set(nil)` flips both endpoints back to 200 and stays there for the rest of the process lifetime — transient ClickHouse blips after boot are reflected in `/ready` only. Resolves #95. + +- **Boot-time schema discovery no longer crashes the binary** (`cmd/wavehouse/main.go`, `internal/api/health.go`, `internal/discovery/discovery.go`, `docs/src/content/docs/api.md`, `docs/src/content/docs/deployment.md`). Previously, any error from the initial `SchemaRegistry.Refresh` — connection-refused, missing database, transient network blip — caused `cmd/wavehouse` to call `os.Exit(1)`. The supervisor would restart the process every ~10s in an unbounded loop, port 8080 never bound, and operators got `connection refused` on probes even when ClickHouse was otherwise healthy. Now the first Refresh failure is non-fatal: a new `api.BootState` is set with the diagnostic, the server still binds `:8080`, and a background goroutine calls the new `SchemaRegistry.RetryRefresh(ctx, 2s, 60s, onAttempt)` with exponential backoff until success or shutdown. While `BootState.Err()` is non-nil, `/health` returns 503 with `{"status":"degraded","error":"…"}` so an operator can `curl /health` to learn why the gateway isn't ready to serve traffic instead of grepping a restart-loop log; `/ready` returns 503 with `{"status":"not ready",…}` for the same reason. Once `BootState.Set(nil)` fires (either from the initial sync `Refresh` or from `SchemaRegistry.RetryRefresh` success), `/health` stays at `200 OK` for the rest of the process lifetime — that flip is sticky and reflects "boot completed once." `/ready`, by contrast, remains conditional on current ClickHouse reachability: it can flip back to `503` on transient runtime blips after boot, since readiness is a probe of "can I serve traffic right now," not "did I ever boot." Resolves #95. ### Removed - **`project-orchestrator.yml` workflow + its three composite-action artifacts** (`.github/workflows/project-orchestrator.yml`, `.github/actions/board-upsert-status/`, `.github/actions/set-linked-issues-status/`, `.github/scripts/board-fetch-item.sh`, `AGENTS.md`, `CHANGELOG.md`): −887 lines net. The orchestrator was the largest single source of cross-trigger complexity on this repo (3-4 workflow_run-chained runs per PR push, `statusCheckRollup` GraphQL perms quirks, integration-token `NONE` for private-org members) for behaviour that is mostly either provided natively by GitHub or a one-click manual operation on a 4-person team. Replaced by: reviewer-assign step in `housekeeping.yml` that fires once on `pull_request_target: opened` / `ready_for_review` (not per-synchronize, so it doesn't re-spam after `dismiss_stale_reviews_on_push`), plus GitHub's native Projects v2 workflows (`Auto-add to project`, `Item added`, `Pull request merged`) configured in the project UI. Trade-offs explicit in the PR body: drafts no longer auto-flip on bot-clean, `CHANGES_REQUESTED` doesn't auto-move the board card, linked-issue card mirroring is dropped. AGENTS.md §"Governance Files" + §"Task Board state machine" + §"Review tooling reference" all rewritten to match. `dependabot-automerge.yml` trimmed in parallel: no more board-upsert step (native handles placement), `PROJECT_BOARD_TOKEN` guard removed (no longer used in this workflow), reviewer list sourced from `board-config.env`'s `ADMINS` via `replace()`, major-bump comment uses the marker-comment upsert pattern from `housekeeping.yml`. diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 7e0a19f1..dcdc6308 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -172,7 +172,9 @@ func run() int { // and retry in the background with exponential backoff. The process // still binds :8080 so operators can `curl /health` instead of grepping // a restart-loop log. Once a Refresh succeeds, bootState flips to nil - // and /health returns 200. + // and /health returns 200. The periodic auto-refresh ticker is started + // only after the first successful Refresh (sync or retry) so it never + // races RetryRefresh on Refresh calls or on bootState writes. bootState := api.NewBootState(nil) refreshInterval := time.Duration(cfg.Schema.RefreshInterval) * time.Second registry := discovery.NewSchemaRegistry(chConn, cfg.ClickHouse.Database, refreshInterval, logger) @@ -190,7 +192,10 @@ func run() int { } logger.Info("schema discovery succeeded after retry, /health now 200") bootState.Set(nil) + go registry.StartAutoRefresh(ctx) }() + } else { + go registry.StartAutoRefresh(ctx) } // State directories — one configurable root, fixed subdir convention. @@ -268,9 +273,6 @@ func run() int { // Hub for streaming fan-out. hub := api.NewHub() - // Start schema auto-refresh. - go registry.StartAutoRefresh(ctx) - // Start policy watch for cluster-wide updates. go policyStore.Watch(ctx) diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md index 33612807..da31141f 100644 --- a/docs/src/content/docs/api.md +++ b/docs/src/content/docs/api.md @@ -90,7 +90,7 @@ Returns `200 OK` once the gateway has discovered ClickHouse table schemas at lea Status code: `503 Service Unavailable` -The boot-degraded response lets an operator `curl /health` to learn why the gateway hasn't bound traffic yet, instead of grepping a restart-loop log. Schema discovery retries with exponential backoff (2s → 60s); once a Refresh succeeds, `/health` flips to `200` and stays there for the rest of the process lifetime — transient ClickHouse blips after that point are reflected in `/ready`, not `/health`. +The boot-degraded response lets an operator `curl /health` to learn why the gateway isn't ready to serve traffic yet, instead of grepping a restart-loop log. The binary is bound on `:8080` and serves diagnostics, but is not yet accepting ingest/query traffic. Schema discovery retries with exponential backoff (2s → 60s); once a Refresh succeeds, `/health` flips to `200` and stays there for the rest of the process lifetime — transient ClickHouse blips after that point are reflected in `/ready`, not `/health`. --- diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 85d4684b..80b1a157 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -479,8 +479,12 @@ func TestRetryRefresh_ReturnsOnContextCancel(t *testing.T) { func TestRetryRefresh_BackoffIsBounded(t *testing.T) { t.Parallel() // Five failures then success; with initial 1ms and max 4ms backoff, - // sleeps are 1, 2, 4, 4, 4 = 15ms total. We assert it stays well under - // the unbounded-doubling worst case (1+2+4+8+16 = 31ms). + // sleeps are 1, 2, 4, 4, 4 = 15ms total. The unbounded-doubling worst + // case would be 1+2+4+8+16 = 31ms. We leave generous headroom on the + // upper bound because shared CI runners can stall the scheduler enough + // to drag a 15ms sleep budget past 100ms; 250ms still catches a real + // unbounded backoff regression (which would balloon by orders of + // magnitude) without flaking on noisy hosts. errs := make([]error, 5) for i := range errs { errs[i] = errors.New("transient") @@ -494,7 +498,7 @@ func TestRetryRefresh_BackoffIsBounded(t *testing.T) { elapsed := time.Since(start) // Lower bound proves we actually slept; upper bound proves capping. assert.GreaterOrEqual(t, elapsed, 10*time.Millisecond) - assert.Less(t, elapsed, 100*time.Millisecond) + assert.Less(t, elapsed, 250*time.Millisecond) } // TestRetryRefresh_NilOnAttemptIsSafe verifies the loop tolerates a nil From e6676acbc5ae11ec30c9eb3fdf8f0756f42c5c16 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:43:26 -0400 Subject: [PATCH 06/13] test(discovery): relax SucceedsOnFirstAttempt wall-clock bound to 250ms MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit Round 2 [Minor] on commit dcb452b: same 100ms-too-tight concern as TestRetryRefresh_BackoffIsBounded, applied here to the happy-path test. Expected wall-clock here is ~0 (no sleeps), but a scheduler stall on a contended CI runner can blow past 100ms even when RetryRefresh behaves correctly. 250ms keeps consistency with the bounded-backoff test and is still orders of magnitude under any real-sleep regression — if the no-sleep short-circuit ever broke, the test would sleep initialBackoff (1h), so the bound only needs to be << 1h, not "as tight as possible." Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/discovery/discovery_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 80b1a157..864b965e 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -414,7 +414,12 @@ func TestRetryRefresh_SucceedsOnFirstAttempt(t *testing.T) { }) require.NoError(t, err) - assert.Less(t, time.Since(start), 100*time.Millisecond, "should not have slept") + // Same 250ms headroom as TestRetryRefresh_BackoffIsBounded — the + // expected wall-clock budget here is ~0 (no sleep at all), but a + // scheduler stall on a contended CI runner can drag a no-sleep test + // past 100ms. 250ms is still orders of magnitude under any real-sleep + // regression (the misbehaviour would sleep `initialBackoff` = 1h). + assert.Less(t, time.Since(start), 250*time.Millisecond, "should not have slept") assert.Equal(t, int32(0), atomic.LoadInt32(&attempts), "onAttempt should only fire on failure") assert.Equal(t, int32(1), conn.calls.Load(), "exactly one Query call expected") } From 48d56d94aa00c61f919f33edbb7fd52f970fe489 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Wed, 13 May 2026 16:48:22 -0400 Subject: [PATCH 07/13] test(discovery): JSON slog handler + WithTimeout ctx in happy-path test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two CodeRabbit Round 3 findings on commit e6676ac: 1) [Major] newFakeRegistry test helper used slog.NewTextHandler — repo coding standard says JSON handler. Switched to slog.NewJSONHandler. (Output is io.Discard either way, so the change is cosmetic — but no reason to leave a TextHandler in the file when JSON is the standard.) 2) [Minor] TestRetryRefresh_SucceedsOnFirstAttempt used context.Background() with time.Hour backoff. A hypothetical regression where RetryRefresh sleeps before checking success would hang the test until the Go framework's 10-minute default timeout. Wrapped with context.WithTimeout(2s) + defer cancel — same bound semantics, but fails fast with a context error if the regression ever happens. `go test -race -count=3 -run TestRetryRefresh ./internal/discovery/...` passes locally. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/discovery/discovery_test.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 864b965e..1c70816c 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -397,7 +397,7 @@ func (*emptyRows) ColumnTypes() []driver.ColumnType { func newFakeRegistry(t *testing.T, errs []error) (*SchemaRegistry, *fakeConn) { t.Helper() conn := &fakeConn{errsThenSuccess: errs} - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + logger := slog.New(slog.NewJSONHandler(io.Discard, nil)) return NewSchemaRegistry(conn, "test", time.Hour, logger), conn } @@ -407,9 +407,17 @@ func TestRetryRefresh_SucceedsOnFirstAttempt(t *testing.T) { t.Parallel() sr, conn := newFakeRegistry(t, nil) + // 2s timeout context bounds a hypothetical regression where RetryRefresh + // starts sleeping before checking success — initialBackoff=time.Hour means + // such a regression would otherwise hang the test until the Go test + // framework's default timeout (10m). With WithTimeout, the test fails fast + // with a context error instead. + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + var attempts int32 start := time.Now() - err := sr.RetryRefresh(context.Background(), time.Hour, time.Hour, func(error) { + err := sr.RetryRefresh(ctx, time.Hour, time.Hour, func(error) { atomic.AddInt32(&attempts, 1) }) require.NoError(t, err) From ec5ea4321965fdbe4a28f0670aecde7c48ca97eb Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 12:07:35 -0400 Subject: [PATCH 08/13] =?UTF-8?q?test:=20close=20the=20contract=20on=20non?= =?UTF-8?q?-fatal=20boot=20=E2=80=94=20chain=20+=20auto-refresh=20+=20prob?= =?UTF-8?q?e=20+=20integration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four test gaps that the original PR left as manual spot-checks: 1. internal/api/boot_chain_test.go — TestBoot_Chain_DegradedThenRecovers. Exercises the wiring between Refresh failure → BootState set → /health 503 with diagnostic → RetryRefresh success → BootState.Set(nil) → /health 200. The pieces are unit-tested individually elsewhere; this nails them down as a pipeline using the production types and a small errsThenSuccessConn fake. 2. internal/discovery/discovery_test.go — TestStartAutoRefresh_LogsAndContinuesOnError. Covers the ERROR-log branch in the ticker loop that prior tests missed: perpetual-error fakeConn + 5ms refreshInterval + assert.Eventually on a mutex-protected slog buffer. Discovery coverage 89.8% → 91.7%. 3. tests/integration/boot_resilience_test.go — TestBootResilience_- StickyHealthVsConditionalReady. Pins the four-row contract matrix the CHANGELOG calls out (boot CH-down, boot CH-up-after-retry, post-boot CH-dies, post-boot CH-back) against a real CH testcontainer that the test stops and starts twice. The post-boot rows are the new behaviour: /health sticky-200 once boot succeeds, /ready conditional on current CH reachability. Needed refreshChAddr() to re-query the host-mapped port after Stop+Start (Docker reassigns). 4. cmd/wavehouse/health_test.go — TestRunHealthCheck_{Success, BootDegraded503, ConnectionRefused, InvalidPortEnv}. The Dockerfile HEALTHCHECK probe is what *reads* the 503 and turns it into a container-unhealthy signal — without these, the operator-visible contract "Docker marks boot-degraded containers unhealthy" was just a claim. Drops the four TODO comments the user added flagging these gaps. The two TODOs in internal/mq/embedded.go for unrelated MQ tunables (over-reservation guard, sync_interval config) move to issue references — filed as #138 (oversubscribe guard) and #139 (fsync tunable). Those are separate features, not test gaps; out of scope for #95. make verify + make test-unit (74.9% combined, all per-package gates green) + make test-integration (21.5%, 13 tests, +my new test) all pass. Closes #95. Co-Authored-By: Claude Opus 4.7 (1M context) --- cmd/wavehouse/health_test.go | 88 +++++++++++ internal/api/boot_chain_test.go | 120 +++++++++++++++ internal/api/health_test.go | 2 +- internal/discovery/discovery_test.go | 66 ++++++++ internal/mq/embedded.go | 3 + tests/integration/boot_resilience_test.go | 176 ++++++++++++++++++++++ 6 files changed, 454 insertions(+), 1 deletion(-) create mode 100644 cmd/wavehouse/health_test.go create mode 100644 internal/api/boot_chain_test.go create mode 100644 tests/integration/boot_resilience_test.go diff --git a/cmd/wavehouse/health_test.go b/cmd/wavehouse/health_test.go new file mode 100644 index 00000000..42612f61 --- /dev/null +++ b/cmd/wavehouse/health_test.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "strconv" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// runHealthCheck is the Dockerfile HEALTHCHECK probe — the bridge between +// the in-process /health endpoint and the docker daemon's container-health +// state. These tests pin the contract that holds the boot-non-fatal promise +// together: when wavehouse is boot-degraded, /health returns 503 and the +// probe must return 1 so Docker marks the container unhealthy. Without this +// coverage, a regression that swallowed non-200 responses would silently +// re-create the "stuck container" failure mode the PR is solving. +// +// Tests don't run in parallel because they all mutate WH_SERVER_PORT. +// t.Setenv handles save/restore. + +// portFromListener pulls the integer port off a net.Listener. We use the +// listener directly (rather than parsing httptest.Server.URL) so the same +// helper works for both started-server cases and the connection-refused +// case where we listen-then-close to claim a known-unused port. +func portFromListener(t *testing.T, ln net.Listener) string { + t.Helper() + addr, ok := ln.Addr().(*net.TCPAddr) + require.True(t, ok, "listener should be a TCP listener") + return strconv.Itoa(addr.Port) +} + +func TestRunHealthCheck_Success(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // The probe targets /health specifically — pin that contract here + // so a refactor that points the probe at /ready (which requires + // ClickHouse) doesn't slip through silently. + assert.Equal(t, "/health", r.URL.Path) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + t.Setenv("WH_SERVER_PORT", portFromListener(t, srv.Listener)) + assert.Equal(t, 0, runHealthCheck(), "200 OK should map to exit 0") +} + +func TestRunHealthCheck_BootDegraded503(t *testing.T) { + // This is the user-facing contract under boot-degraded state: the + // in-process /health returns 503 with a diagnostic body, the probe + // sees the non-200, returns 1, and Docker's HEALTHCHECK marks the + // container unhealthy. The K8s startupProbe path in deployment.md + // relies on exactly this signal. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"status":"degraded","error":"schema discovery: connection refused"}`)) + })) + defer srv.Close() + + t.Setenv("WH_SERVER_PORT", portFromListener(t, srv.Listener)) + assert.Equal(t, 1, runHealthCheck(), "503 should map to exit 1 — Docker marks unhealthy") +} + +func TestRunHealthCheck_ConnectionRefused(t *testing.T) { + // Wavehouse hasn't bound :8080 yet (e.g., still in the pre-listen + // phase of `run()`, or crashed entirely). Bind+close a TCP listener + // to claim a port that's now guaranteed unused, so the probe's GET + // gets a real connect-refused rather than a slow timeout. + var lc net.ListenConfig + ln, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + port := portFromListener(t, ln) + require.NoError(t, ln.Close()) + + t.Setenv("WH_SERVER_PORT", port) + assert.Equal(t, 1, runHealthCheck(), "connection refused should map to exit 1") +} + +func TestRunHealthCheck_InvalidPortEnv(t *testing.T) { + // Garbage env var should fail fast at the parse step, not propagate + // a confusing dial error. Catches the case where an operator + // fat-fingers WH_SERVER_PORT in a deployment manifest. + t.Setenv("WH_SERVER_PORT", "not-a-number") + assert.Equal(t, 1, runHealthCheck(), "non-integer WH_SERVER_PORT should exit 1") +} diff --git a/internal/api/boot_chain_test.go b/internal/api/boot_chain_test.go new file mode 100644 index 00000000..bf9f017b --- /dev/null +++ b/internal/api/boot_chain_test.go @@ -0,0 +1,120 @@ +package api + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Wave-RF/WaveHouse/internal/discovery" +) + +// errsThenSuccessConn drives the boot Refresh/RetryRefresh chain in tests. +// Each Query returns the next entry of errs; once exhausted, queries return +// an empty rowset so Refresh succeeds with zero tables. driver.Conn is +// embedded nil — none of its other methods are reached on the discovery +// path, so this minimal stub is sufficient and immune to interface drift. +type errsThenSuccessConn struct { + driver.Conn + errs []error + calls atomic.Int32 +} + +func (c *errsThenSuccessConn) Query(_ context.Context, _ string, _ ...any) (driver.Rows, error) { + n := c.calls.Add(1) + if int(n) <= len(c.errs) { + return nil, c.errs[n-1] + } + return &chainEmptyRows{}, nil +} + +type chainEmptyRows struct{ driver.Rows } + +func (*chainEmptyRows) Next() bool { return false } +func (*chainEmptyRows) Close() error { return nil } +func (*chainEmptyRows) Err() error { return nil } +func (*chainEmptyRows) ColumnTypes() []driver.ColumnType { return nil } + +// TestBoot_Chain_DegradedThenRecovers nails down the full sequence the PR +// adds: the initial Refresh on a stopped/unreachable ClickHouse fails → +// BootState carries the diagnostic → /health returns 503 with the wrapped +// error → RetryRefresh keeps trying with backoff → success clears BootState +// → /health flips to 200. +// +// The pieces are unit-tested in isolation (BootState set/get in this file, +// RetryRefresh happy/retry/cancel paths in discovery_test.go, Liveness 503 +// branch in TestHealth_Liveness_BootDegraded). This test pins them as a +// working pipeline using the production types — a refactor that breaks the +// wiring between Refresh failure → BootState → Liveness 503 → RetryRefresh +// success → BootState.Set(nil) → Liveness 200 fails here even if every +// individual unit test still passes. +func TestBoot_Chain_DegradedThenRecovers(t *testing.T) { + t.Parallel() + + connRefused := errors.New("dial tcp 127.0.0.1:9000: connect: connection refused") + dbMissing := errors.New("code: 81, message: Database wavehouse does not exist") + // Three failures then success — matches the realistic case where CH + // comes up partway through the retry backoff. + conn := &errsThenSuccessConn{errs: []error{connRefused, connRefused, dbMissing}} + + logger := slog.New(slog.NewJSONHandler(io.Discard, nil)) + registry := discovery.NewSchemaRegistry(conn, "test", time.Hour, logger) + + // Phase 0 — synchronous boot Refresh fails. main.go records the + // diagnostic in BootState and proceeds with the retry loop in a + // goroutine; we drive both inline here for determinism. + bootState := NewBootState(nil) + err := registry.Refresh(context.Background()) + require.Error(t, err, "Refresh against unreachable CH must fail") + bootState.Set(fmt.Errorf("schema discovery: %w", err)) + + handler := NewHealthHandler(nil) + handler.Boot = bootState + + // Phase 1 — /health surfaces the diagnostic verbatim with 503. This is + // the operator-facing contract: `curl /health` answers "why can't this + // process serve traffic" instead of the operator having to grep a + // restart-loop log. + rec := httptest.NewRecorder() + handler.Liveness(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/health", nil)) + assert.Equal(t, http.StatusServiceUnavailable, rec.Code) + assert.Contains(t, rec.Body.String(), `"status":"degraded"`) + assert.Contains(t, rec.Body.String(), "connection refused") + + // Phase 2 — background retry loop drives Refresh to eventual success. + // Tight bounds keep the test fast; the retry-loop unit tests cover the + // actual backoff math. + retryCtx, retryCancel := context.WithTimeout(context.Background(), 2*time.Second) + defer retryCancel() + retryErr := registry.RetryRefresh(retryCtx, time.Millisecond, 10*time.Millisecond, func(attemptErr error) { + bootState.Set(fmt.Errorf("schema discovery: %w", attemptErr)) + }) + require.NoError(t, retryErr, "retry loop should succeed on the 4th attempt") + bootState.Set(nil) + + // Phase 3 — /health flips to 200 once BootState is cleared. This is + // the sticky-from-here invariant the CHANGELOG calls out: BootState + // won't be touched again for the rest of the process lifetime, so + // /health remains 200 even if ClickHouse goes unreachable later + // (that case is reflected in /ready, not /health — covered by the + // integration test in tests/integration/boot_resilience_test.go). + rec = httptest.NewRecorder() + handler.Liveness(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/health", nil)) + assert.Equal(t, http.StatusOK, rec.Code) + assert.Contains(t, rec.Body.String(), `"status":"ok"`) + + // Sanity: exactly four Query calls — the initial sync Refresh (1) plus + // the retry loop's three attempts (2,3 fail; 4 succeeds). An off-by-one + // in the loop's success short-circuit would show up here. + assert.Equal(t, int32(4), conn.calls.Load(), "expected 4 Query calls total") +} diff --git a/internal/api/health_test.go b/internal/api/health_test.go index 7e632ee5..bb3999e3 100644 --- a/internal/api/health_test.go +++ b/internal/api/health_test.go @@ -100,7 +100,7 @@ func TestBootState_SetAndGet(t *testing.T) { func TestHealth_Liveness_BootDegraded(t *testing.T) { t.Parallel() // When BootState reports a non-nil error, Liveness must return 503 with - // the diagnostic in the JSON body. This is the behaviour the gateway + // the diagnostic in the JSON body. This is the behavior the gateway // relies on so an operator can `curl /health` during a boot-time // ClickHouse outage instead of grepping a restart-loop log. h := NewHealthHandler(nil) diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 1c70816c..68f5a0c1 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -1,10 +1,13 @@ package discovery import ( + "bytes" "context" "errors" "io" "log/slog" + "strings" + "sync" "sync/atomic" "testing" "time" @@ -579,3 +582,66 @@ func TestStartAutoRefresh_ExitsOnContextCancel(t *testing.T) { t.Fatal("StartAutoRefresh did not return after ctx cancel") } } + +// concurrentBuffer wraps bytes.Buffer with a mutex so the test goroutine can +// read the buffer while slog's JSON handler writes to it from another. slog +// serialises its own writes via an internal mutex, but the test's reads sit +// outside that mutex — without external synchronisation, `go test -race` +// reports the buf access as a data race. +type concurrentBuffer struct { + mu sync.Mutex + b bytes.Buffer +} + +func (c *concurrentBuffer) Write(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.b.Write(p) +} + +func (c *concurrentBuffer) String() string { + c.mu.Lock() + defer c.mu.Unlock() + return c.b.String() +} + +// TestStartAutoRefresh_LogsAndContinuesOnError covers the error branch in +// StartAutoRefresh's ticker loop: a failed Refresh logs an ERROR line and +// the loop keeps going. Operators rely on this so transient ClickHouse +// blips after boot don't kill the auto-refresh goroutine — the schema +// cache stays warm until the next successful tick. +// +// Uses a perpetual-error fake conn + a 5ms refresh interval to trigger the +// branch in milliseconds, then asserts the log line is present via +// assert.Eventually (no time.Sleep-based scheduling assumption). +func TestStartAutoRefresh_LogsAndContinuesOnError(t *testing.T) { + t.Parallel() + errs := make([]error, 50) + for i := range errs { + errs[i] = errors.New("transient") + } + conn := &fakeConn{errsThenSuccess: errs} + + var buf concurrentBuffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + sr := NewSchemaRegistry(conn, "test", 5*time.Millisecond, logger) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + sr.StartAutoRefresh(ctx) + close(done) + }() + + assert.Eventually(t, func() bool { + return strings.Contains(buf.String(), "schema auto-refresh failed") + }, 2*time.Second, 5*time.Millisecond, "expected auto-refresh failure log line within 2s") + + cancel() + <-done + + // Sanity: the ticker actually fired Refresh at least once. Without + // this, an off-by-one that skipped the first tick would silently let + // the test pass off a stale buffer assertion on a never-run loop. + assert.Greater(t, conn.calls.Load(), int32(0), "ticker never fired Refresh") +} diff --git a/internal/mq/embedded.go b/internal/mq/embedded.go index a9bcbb4d..41ecfbbd 100644 --- a/internal/mq/embedded.go +++ b/internal/mq/embedded.go @@ -57,6 +57,9 @@ func NewEmbedded(storeDir string, maxBytes int64, logger ...*slog.Logger) (*Embe l = logger[0] } + // TODO(#138): guard against too much space reserved (validate maxBytes vs statfs(storeDir)) + // TODO(#139): allow configuring JetStream sync_interval (throughput vs durability tradeoff) + opts := &natsserver.Options{ DontListen: true, JetStream: true, diff --git a/tests/integration/boot_resilience_test.go b/tests/integration/boot_resilience_test.go new file mode 100644 index 00000000..35dcef8e --- /dev/null +++ b/tests/integration/boot_resilience_test.go @@ -0,0 +1,176 @@ +//go:build integration + +package tests + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Wave-RF/WaveHouse/internal/api" + "github.com/Wave-RF/WaveHouse/internal/discovery" + "github.com/Wave-RF/WaveHouse/internal/testutil" +) + +// TestBootResilience_StickyHealthVsConditionalReady exercises the full +// post-#95 contract matrix against a real ClickHouse testcontainer that we +// stop/start mid-test: +// +// | State | /health | /ready | +// |--------------------------|---------|--------| +// | Boot, CH down | 503 | 503 | +// | Boot, CH up after retry | 200 | 200 | +// | Post-boot, CH dies | 200 ★ | 503 | +// | Post-boot, CH back | 200 | 200 | +// +// ★ — the sticky-/health invariant the PR adds, where /health stays at +// "boot completed once" rather than reflecting current ClickHouse state. +// The unit chain test (internal/api/boot_chain_test.go) pins the same +// wiring with a fake conn; this test additionally pins it against a real +// CH driver dial AND covers the sticky-vs-conditional dichotomy that the +// unit test can't exercise (it has no concept of "CH dies post-boot"). +// +// Uses its own per-test CH testcontainer rather than sharedEnv — the +// shared env assumes CH stays up for the duration of every test in this +// package, which is exactly the assumption this test needs to violate. +func TestBootResilience_StickyHealthVsConditionalReady(t *testing.T) { + ctx := context.Background() + logger := testutil.NopLogger() + + ch, err := startClickHouse(ctx) + require.NoError(t, err, "starting initial CH container") + t.Cleanup(func() { + if ch.conn != nil { + _ = ch.conn.Close() + } + // Always terminate, even after intermediate Stop()s — the test + // may have stopped the container as part of the matrix below. + _ = ch.container.Terminate(context.Background()) + }) + + // Phase 0 — drop CH to its stopped state so the first Refresh sees a + // real conn-refused, not a half-ready boot. + stopTimeout := 10 * time.Second + require.NoError(t, ch.container.Stop(ctx, &stopTimeout), "stop CH before boot phase") + + // Close + reopen the driver. The original conn was Open'd while CH + // was up so it has a healthy idle pool; using it now would race + // against the driver discarding the dead pool entries. Fresh conn + // gives us a clean conn-refused on the first Refresh. + _ = ch.conn.Close() + ch.conn, err = openDriver(ch.nativeAddr()) + require.NoError(t, err, "reopen driver against stopped CH") + + bootState := api.NewBootState(nil) + registry := discovery.NewSchemaRegistry(ch.conn, testCHDatabase, time.Minute, logger) + + // === Row 1: Boot, CH down === + err = registry.Refresh(ctx) + require.Error(t, err, "Refresh against stopped CH must fail") + bootState.Set(fmt.Errorf("schema discovery: %w", err)) + + h := api.NewHealthHandler(ch.conn) + h.Boot = bootState + + assertHealth(t, "/health", h.Liveness, http.StatusServiceUnavailable, `"status":"degraded"`) + assertHealth(t, "/ready", h.Readiness, http.StatusServiceUnavailable, `"status":"not ready"`) + + // Restart CH and refresh the cached mapped port — Docker may reassign + // the host port on Stop+Start, so the chInstance.nativePort cached at + // initial startClickHouse() can be stale. Rebuild the driver against + // the live port before waiting for readiness. + require.NoError(t, ch.container.Start(ctx), "restart CH for recovery phase") + require.NoError(t, refreshChAddr(ctx, ch), "refresh mapped port after restart") + _ = ch.conn.Close() + ch.conn, err = openDriver(ch.nativeAddr()) + require.NoError(t, err, "reopen driver against restarted CH") + registry = discovery.NewSchemaRegistry(ch.conn, testCHDatabase, time.Minute, logger) + h.CHConn = ch.conn + require.NoError(t, waitForNativeReady(ctx, ch.conn, 30*time.Second), "CH native should be ready after restart") + + retryCtx, retryCancel := context.WithTimeout(ctx, 30*time.Second) + defer retryCancel() + require.NoError(t, registry.RetryRefresh(retryCtx, 250*time.Millisecond, 2*time.Second, func(e error) { + bootState.Set(fmt.Errorf("schema discovery: %w", e)) + }), "RetryRefresh should succeed once CH is back") + bootState.Set(nil) + + // === Row 2: Boot, CH up after retry === + assertHealth(t, "/health", h.Liveness, http.StatusOK, `"status":"ok"`) + assertHealth(t, "/ready", h.Readiness, http.StatusOK, `"status":"ready"`) + + // Now simulate "CH dies post-boot" — the runtime side of the contract. + require.NoError(t, ch.container.Stop(ctx, &stopTimeout), "stop CH for post-boot outage") + + // === Row 3: Post-boot, CH dies === + // /health must stay 200 — BootState is nil and never gets touched + // again, that's the sticky invariant. /ready must drop to 503 because + // the readiness handler still pings CH on every call. + assertHealth(t, "/health", h.Liveness, http.StatusOK, `"status":"ok"`) + assertHealth(t, "/ready", h.Readiness, http.StatusServiceUnavailable, `"status":"not ready"`) + + // Restart CH a final time. Same port-may-have-changed dance — refresh + // the mapped port, rebuild the driver, point the health handler at + // the new conn. /ready is the only assertion here, so registry is + // not re-wired (it has cached schemas from the earlier success and + // the StartAutoRefresh ticker isn't running in this test). + require.NoError(t, ch.container.Start(ctx), "restart CH for final recovery") + require.NoError(t, refreshChAddr(ctx, ch), "refresh mapped port after final restart") + _ = ch.conn.Close() + ch.conn, err = openDriver(ch.nativeAddr()) + require.NoError(t, err, "reopen driver against final restart") + h.CHConn = ch.conn + require.NoError(t, waitForNativeReady(ctx, ch.conn, 30*time.Second), "CH native should be ready after second restart") + + // === Row 4: Post-boot, CH back === + assertHealth(t, "/health", h.Liveness, http.StatusOK, `"status":"ok"`) + assertHealth(t, "/ready", h.Readiness, http.StatusOK, `"status":"ready"`) +} + +// assertHealth invokes the given handler with a GET to path and asserts the +// status code + body substring. Wraps the boilerplate four lines that show +// up at every row of the contract matrix above. +func assertHealth(t *testing.T, path string, fn http.HandlerFunc, wantStatus int, wantSubstr string) { + t.Helper() + rec := httptest.NewRecorder() + fn(rec, httptest.NewRequestWithContext(context.Background(), http.MethodGet, path, nil)) + assert.Equal(t, wantStatus, rec.Code, "%s status", path) + assert.Contains(t, rec.Body.String(), wantSubstr, "%s body", path) +} + +// openDriver mirrors the chConn construction inside startClickHouse so the +// boot test can rebuild the driver after a Stop without duplicating the +// option block. +func openDriver(addr string) (driver.Conn, error) { + return clickhouse.Open(&clickhouse.Options{ + Addr: []string{addr}, + Auth: clickhouse.Auth{ + Database: testCHDatabase, + Username: testCHUser, + Password: testCHPassword, + }, + }) +} + +// refreshChAddr re-queries Docker for the container's currently-mapped +// 9000/tcp host port and updates ch.nativePort in place. Necessary after +// Stop+Start cycles: Docker preserves the container ID but reassigns the +// host port (the binding was released when the container stopped, picked +// up by something else, etc.). The cached value from initial startup +// becomes stale. +func refreshChAddr(ctx context.Context, ch *chInstance) error { + p, err := ch.container.MappedPort(ctx, "9000") + if err != nil { + return fmt.Errorf("mapped port: %w", err) + } + ch.nativePort = p.Port() + return nil +} From 3d9f964d1d5885d4729637285dd3e7fbaee2db8c Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 12:14:29 -0400 Subject: [PATCH 09/13] test: address CodeRabbit round 5 on ec5ea43 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two minor findings, both quick-wins: 1) [Minor] internal/discovery/discovery_test.go:641 — the new TestStartAutoRefresh_LogsAndContinuesOnError used a raw `<-done` after cancel(), which can hang if StartAutoRefresh ever fails to exit. Wrapped with the same select+2s-timeout the existing TestStartAutoRefresh_ExitsOnContextCancel uses — fails fast on hangs instead of waiting for the Go test framework's 10-minute default. 2) [Minor] internal/api/health_test.go:163 — TestHealth_Readiness_BootDegradedReports503 was missing the Content-Type/X-Content-Type-Options header assertions that the other 503-path tests in the file pin. Added them so a future refactor that moves header setup into the success branch fails here too. Same rationale I argued for in the round-1 pushback on the testutil-helpers suggestion: header coverage on error paths is deliberate. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/api/health_test.go | 2 ++ internal/discovery/discovery_test.go | 7 ++++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/internal/api/health_test.go b/internal/api/health_test.go index bb3999e3..1ac18e9d 100644 --- a/internal/api/health_test.go +++ b/internal/api/health_test.go @@ -156,6 +156,8 @@ func TestHealth_Readiness_BootDegradedReports503(t *testing.T) { h.Readiness(w, r) assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Equal(t, "application/json", w.Header().Get("Content-Type")) + assert.Equal(t, "nosniff", w.Header().Get("X-Content-Type-Options")) var resp map[string]string require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) assert.Equal(t, "not ready", resp["status"]) diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 68f5a0c1..5e9464f5 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -638,7 +638,12 @@ func TestStartAutoRefresh_LogsAndContinuesOnError(t *testing.T) { }, 2*time.Second, 5*time.Millisecond, "expected auto-refresh failure log line within 2s") cancel() - <-done + select { + case <-done: + // expected — ticker observed ctx.Done() and exited + case <-time.After(2 * time.Second): + t.Fatal("StartAutoRefresh did not return after ctx cancel") + } // Sanity: the ticker actually fired Refresh at least once. Without // this, an off-by-one that skipped the first tick would silently let From 11a3b4db43507e6c9bb851ea8282d9918c987c54 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 12:49:55 -0400 Subject: [PATCH 10/13] build: include ./cmd/... in test-unit + file-level cov excludes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The new cmd/wavehouse/health_test.go was running locally (go test ./cmd/wavehouse/...) but NOT under make test-unit, because the unit target was hardcoded to ./internal/... — so the regression-protection value of the new probe tests wasn't being enforced in CI. The cmd/wavehouse line was simply missing from the coverage breakdown. Three coordinated changes: 1. Makefile test-unit invocation widened to ./internal/... ./cmd/.... Tests now actually run in CI; cmd/wavehouse appears in the per- package breakdown. 2. scripts/cov/main.go's exclude-matching now checks both pkg+"/" (existing — directory-level) and the file path (new — file-level). Single-line semantic widening: existing patterns keep working unchanged, new patterns like ^cmd/wavehouse/main\.go$ become possible. 3. .testcoverage.yml adds ^cmd/wavehouse/main\.go$ to exclude.paths. main.go is 400+ lines of service wiring that's only meaningfully exercised by the e2e suite (bin/wavehouse-cov against real ClickHouse+NATS) — including it in the unit coverage average dragged the gate from 74.9% to 68.8% even though zero new unit-testable code regressed. health.go in the same package stays counted (91.7% from the new tests). Net effect on make test-unit output: cmd/wavehouse: 91.7% appears in the breakdown; total 75.2% (vs 74.9% prior), gate passes. Verifies visually that the new probe tests are running + protecting a contract. Co-Authored-By: Claude Opus 4.7 (1M context) --- .testcoverage.yml | 7 +++++++ Makefile | 2 +- internal/mq/embedded.go | 3 --- scripts/cov/main.go | 13 ++++++++++--- 4 files changed, 18 insertions(+), 7 deletions(-) diff --git a/.testcoverage.yml b/.testcoverage.yml index 40fa951c..ca1449e9 100644 --- a/.testcoverage.yml +++ b/.testcoverage.yml @@ -41,3 +41,10 @@ exclude: # aren't part of the shipped binary; they show up in `-coverpkg=./...` # but never get exercised by tests. - ^scripts/ + # cmd/wavehouse/main.go is the binary entry point — 400+ lines of + # service wiring that's only meaningfully exercised by the e2e suite + # (which runs bin/wavehouse-cov against a real ClickHouse + NATS). + # health.go in the same package IS unit-testable (httptest server + + # WH_SERVER_PORT) and stays in coverage. File-level exclusion requires + # the path-aware match added to scripts/cov/main.go. + - ^cmd/wavehouse/main\.go$ diff --git a/Makefile b/Makefile index 1f7b28bc..9630b8ed 100644 --- a/Makefile +++ b/Makefile @@ -406,7 +406,7 @@ test-unit: go-mod-download ## Run Go unit tests + render coverage + gate thresho @printf "$(CYAN)==> Running Unit Tests...$(RESET)\n" @rm -rf $(COV_UNIT)/data && mkdir -p $(COV_UNIT)/data @GOCOVERDIR="$(CURDIR)/$(COV_UNIT)/data" go tool gotestsum --format $(GOTESTSUM_FMT) -- \ - -tags="$(TAGS)" -cover -race ./internal/... $(ARGS) \ + -tags="$(TAGS)" -cover -race ./internal/... ./cmd/... $(ARGS) \ -args -test.gocoverdir="$(CURDIR)/$(COV_UNIT)/data" @go run ./scripts/cov render unit diff --git a/internal/mq/embedded.go b/internal/mq/embedded.go index 41ecfbbd..a9bcbb4d 100644 --- a/internal/mq/embedded.go +++ b/internal/mq/embedded.go @@ -57,9 +57,6 @@ func NewEmbedded(storeDir string, maxBytes int64, logger ...*slog.Logger) (*Embe l = logger[0] } - // TODO(#138): guard against too much space reserved (validate maxBytes vs statfs(storeDir)) - // TODO(#139): allow configuring JetStream sync_interval (throughput vs durability tradeoff) - opts := &natsserver.Options{ DontListen: true, JetStream: true, diff --git a/scripts/cov/main.go b/scripts/cov/main.go index f392c6bc..fe9dc658 100644 --- a/scripts/cov/main.go +++ b/scripts/cov/main.go @@ -258,9 +258,16 @@ func parseCoverage(profile string, c *config) (rows []pkgRow, total, covered int if c.LocalPrefix != "" && strings.HasPrefix(pkg, c.LocalPrefix+"/") { pkg = pkg[len(c.LocalPrefix)+1:] } - // Match exclusions against pkg+"/" so a pattern like - // ^internal/testutil/ matches a bare-directory pkg string. - if matchesAny(excludes, pkg+"/") { + shortPath := path + if c.LocalPrefix != "" && strings.HasPrefix(shortPath, c.LocalPrefix+"/") { + shortPath = shortPath[len(c.LocalPrefix)+1:] + } + // Match exclusions against either pkg+"/" (directory-level) or + // shortPath (file-level). Directory patterns like ^internal/testutil/ + // match the former; single-file patterns like ^cmd/wavehouse/main\.go$ + // match the latter, useful for excluding binary entry points that + // can't be unit-tested without dragging the suite gate. + if matchesAny(excludes, pkg+"/") || matchesAny(excludes, shortPath) { continue } stmts, _ := strconv.Atoi(f[1]) From f8542745fd22fb76416b6dd6254a5c689e271976 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 14:19:39 -0400 Subject: [PATCH 11/13] build(cov): per-suite excludes + inline threshold gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue: my previous commit added \`^cmd/wavehouse/main\.go\$\` to \`exclude.paths\` in \`.testcoverage.yml\` so the unit gate would pass with the new \`./cmd/...\` test scope. But \`exclude.paths\` is global — applied to every render including the merged total — so I'd also hidden main.go's e2e-derived coverage from the project-wide 80% gate. Net effect: main.go was effectively untracked everywhere, when what we actually wanted was "untracked by unit, tracked by e2e and merged". Two coordinated changes: 1. \`.testcoverage.yml\` schema gets a new \`exclude.per-suite\` section. Patterns under \`per-suite.\` apply on top of global \`paths\` but only when rendering that suite's profile — they're intentionally NOT applied when computing the merged total. Moved \`^cmd/wavehouse/main\.go\$\` from global to \`per-suite.unit\` with a comment cross-referencing #140 (the deferred main.go refactor that would let us drop the exclude entirely). 2. \`scripts/cov/main.go\` grows a \`config.excludesFor(suite)\` helper that returns \`Paths\` plus that suite's \`PerSuite\` list (or just \`Paths\` when \`suite == ""\` / "total"). \`parseCoverage\` now takes the exclude list explicitly so the renderSuite and merge call sites can pass different lists. While I was in there, also replaced the external \`go-test-coverage --threshold-total=N\` subprocess gate with an inline integer-arithmetic check (\`meetsThreshold\`): - That tool predates per-suite excludes and re-parses the YAML with global-only semantics, so its verdict diverged from our own breakdown for any suite that had per-suite excludes (saw unit pass at 75.1% in our breakdown but the external gate fail at 68.8% because it still counted main.go). - The inline check uses the same exclude set our breakdown does, computed once. One source of truth, no subprocess. - \`covered*100 >= total*threshold\` avoids float-boundary rounding (e.g., 69.99999% should fail the 70% gate; with floats we'd have to decide which side rounds). Verified behaviour: - \`make test-unit\`: cmd/wavehouse 91.7% (health.go), main.go excluded, suite total 75.1%, gate passes. - \`make test-integration\`: 21.5%, gate passes. main.go absent from the integration profile entirely — not a per-suite-exclude effect; cmd/wavehouse is package main and isn't imported by the integration test binary, so Go's compiler never instruments it. This is fundamental to the test layout, not a config issue. - \`make cov\` (unit + integration + e2e merged): cmd/wavehouse 62.1% (health.go via unit + main.go via e2e), total 81.8%, gate passes. Closes the loop on what the previous commit was *trying* to do. Co-Authored-By: Claude Opus 4.7 (1M context) --- .testcoverage.yml | 27 ++++++++++---- scripts/cov/main.go | 87 ++++++++++++++++++++++++++++++++++----------- 2 files changed, 86 insertions(+), 28 deletions(-) diff --git a/.testcoverage.yml b/.testcoverage.yml index ca1449e9..18892eed 100644 --- a/.testcoverage.yml +++ b/.testcoverage.yml @@ -31,6 +31,8 @@ suites: sdk: 50 # statements (vitest gates via --coverage.thresholds.statements) exclude: + # Global excludes apply to every suite render AND to the merged total. + # Use for paths that aren't meaningful coverage anywhere. paths: # internal/testutil/ contains shared test helpers (mocks, JWT builders, # HTTP assertions). It is imported only from *_test.go files, never from @@ -41,10 +43,21 @@ exclude: # aren't part of the shipped binary; they show up in `-coverpkg=./...` # but never get exercised by tests. - ^scripts/ - # cmd/wavehouse/main.go is the binary entry point — 400+ lines of - # service wiring that's only meaningfully exercised by the e2e suite - # (which runs bin/wavehouse-cov against a real ClickHouse + NATS). - # health.go in the same package IS unit-testable (httptest server + - # WH_SERVER_PORT) and stays in coverage. File-level exclusion requires - # the path-aware match added to scripts/cov/main.go. - - ^cmd/wavehouse/main\.go$ + + # Per-suite excludes layer on top of `paths` but only for that suite's + # render. Crucially, they do NOT apply to the merged total — so a file + # excluded from `unit` here still counts toward the project-wide gate + # via whichever suite(s) actually do cover it. See scripts/cov/main.go + # config.excludesFor for the wiring. + per-suite: + unit: + # cmd/wavehouse/main.go is the binary entry point — 400+ lines of + # service wiring that's only meaningfully exercised by the e2e suite + # (which runs bin/wavehouse-cov against a real ClickHouse + NATS). + # Excluding from `unit` keeps that gate from being dragged down by + # 0% coverage on an untestable wiring file. The merged total still + # counts e2e's coverage of this file. health.go in the same package + # IS unit-testable (httptest server + WH_SERVER_PORT) and stays in + # coverage everywhere — see issue #140 for the deferred refactor + # that would shrink main.go enough to drop this exclude entirely. + - ^cmd/wavehouse/main\.go$ diff --git a/scripts/cov/main.go b/scripts/cov/main.go index fe9dc658..54a30577 100644 --- a/scripts/cov/main.go +++ b/scripts/cov/main.go @@ -49,10 +49,33 @@ type config struct { } `yaml:"threshold"` Suites map[string]int `yaml:"suites"` Exclude struct { + // Paths applies to every per-suite render AND to the merged total. + // Use for paths that aren't meaningful coverage anywhere (e.g., + // internal/testutil/, tests/, scripts/). Paths []string `yaml:"paths"` + // PerSuite applies on top of Paths but only when rendering the + // named suite's profile — not when computing the merged total. + // Use for files that legitimately can't be covered by one suite + // (e.g., cmd/wavehouse/main.go is untestable by unit but exercised + // by e2e); excluding from `unit` keeps that suite's gate clean + // without hiding the file's e2e-derived coverage from the merged + // total. + PerSuite map[string][]string `yaml:"per-suite"` } `yaml:"exclude"` } +// excludesFor returns the regex patterns that apply when rendering the +// given suite: global Paths plus that suite's PerSuite list (if any). +// Pass an empty suite name (or "total") to get only global excludes — +// what the merged-total render uses. +func (c *config) excludesFor(suite string) []string { + out := append([]string(nil), c.Exclude.Paths...) + if suite != "" && suite != "total" { + out = append(out, c.Exclude.PerSuite[suite]...) + } + return out +} + const ( cyan = "\x1b[36m" green = "\x1b[32m" @@ -141,7 +164,7 @@ func renderSuite(c *config, suite string) error { return err } threshold := thresholdFor(c, suite) - rows, total, covered, err := parseCoverage(profile, c) + rows, total, covered, err := parseCoverage(profile, c, c.excludesFor(suite)) if err != nil { return err } @@ -150,9 +173,10 @@ func renderSuite(c *config, suite string) error { printBreakdown(rows, threshold) fmt.Printf("\n HTML: %s\n\n", htmlOut) - if err := gate(profile, threshold); err != nil { - fmt.Fprintf(os.Stderr, "%s==> %s gate FAILED (below %d%%)%s\n", red, suite, threshold, reset) - return err + if !meetsThreshold(covered, total, threshold) { + fmt.Fprintf(os.Stderr, "%s==> %s gate FAILED (%s < %d%%)%s\n", + red, suite, formatPct(covered, total), threshold, reset) + return fmt.Errorf("%s gate failed: %s < %d%%", suite, formatPct(covered, total), threshold) } fmt.Printf("%s==> %s gate passed (≥ %d%%)%s\n", green, suite, threshold, reset) return nil @@ -201,7 +225,11 @@ func merge(c *config) error { } threshold := c.Threshold.Total - rows, _, _, err := parseCoverage(profile, c) + // Merged total uses only global excludes — per-suite excludes are + // intentionally NOT applied here, so a file that's untestable by one + // suite (e.g., cmd/wavehouse/main.go vs unit) still counts toward the + // project-wide gate via the suite(s) that DO cover it (e2e). + rows, total, covered, err := parseCoverage(profile, c, c.excludesFor("")) if err != nil { return err } @@ -210,11 +238,13 @@ func merge(c *config) error { fmt.Printf("\n HTML: %s\n\n", htmlOut) fmt.Printf("%s==> Combined coverage gate (threshold %d%%):%s\n", cyan, threshold, reset) - if err := gate(profile, threshold); err != nil { - fmt.Fprintf(os.Stderr, "%s==> Total gate FAILED%s\n", red, reset) - return err + if !meetsThreshold(covered, total, threshold) { + fmt.Fprintf(os.Stderr, "%s==> Total gate FAILED (%s < %d%%)%s\n", + red, formatPct(covered, total), threshold, reset) + return fmt.Errorf("total gate failed: %s < %d%%", formatPct(covered, total), threshold) } - fmt.Printf("%s==> Total gate passed%s\n", green, reset) + fmt.Printf("%s==> Total gate passed (%s ≥ %d%%)%s\n", + green, formatPct(covered, total), threshold, reset) return nil } @@ -224,17 +254,23 @@ type pkgRow struct { } // parseCoverage reads a textfmt coverage profile, groups statements by -// package, applies local-prefix stripping and exclude.paths from the config, +// package, applies local-prefix stripping and the supplied exclude patterns, // and returns the per-package rows plus aggregate totals. The aggregates // match what go-test-coverage's gate evaluates, so callers can render // headers that line up with its "Total test coverage" output. -func parseCoverage(profile string, c *config) (rows []pkgRow, total, covered int, err error) { +// +// Callers control which excludes apply by passing the patterns directly +// (see config.excludesFor). This is the seam that makes per-suite excludes +// possible: a unit-only exclude lives in c.Exclude.PerSuite["unit"], the +// renderSuite call passes it in, and merge() omits it so the file still +// shows up in the merged total when e2e covers it. +func parseCoverage(profile string, c *config, excludePatterns []string) (rows []pkgRow, total, covered int, err error) { // #nosec G304,G703 — profile is a coverage.txt path we rendered ourselves. raw, err := os.ReadFile(profile) if err != nil { return nil, 0, 0, err } - excludes := compileRegexes(c.Exclude.Paths) + excludes := compileRegexes(excludePatterns) pkgs := map[string]*pkgRow{} for i, line := range strings.Split(string(raw), "\n") { if i == 0 || line == "" { @@ -376,21 +412,30 @@ func suitePct(c *config, suite string) string { } profile = tmp.Name() } - _, total, covered, err := parseCoverage(profile, c) + _, total, covered, err := parseCoverage(profile, c, c.excludesFor(suite)) if err != nil || total == 0 { return "?" } return formatPct(covered, total) } -// gate runs go-test-coverage with the given total threshold. Exclusions -// (testutil, tests/) come from the same .testcoverage.yml. -func gate(profile string, threshold int) error { - return sh("go", "tool", "go-test-coverage", - "--config="+configPath, - "--profile="+profile, - "--threshold-total="+strconv.Itoa(threshold), - ) +// meetsThreshold returns true iff covered/total * 100 ≥ threshold, using +// integer arithmetic to avoid floating-point rounding at the boundary +// (e.g., 69.99999% should fail the 70% gate; covered*100 >= total*threshold +// is exact). Empty profile (total == 0) is treated as passing — same +// semantics as `go test -cover` reporting "no test files" rather than 0%. +// +// Replaces the previous `go-test-coverage --threshold-total=N` subprocess +// call. That tool predated this PR's per-suite excludes (.testcoverage.yml +// `exclude.per-suite`) and re-parsed the YAML with global-only semantics, +// which made the gate disagree with our breakdown for any suite that had +// per-suite excludes. Doing the check inline guarantees gate and breakdown +// use the same exclude set, computed once. +func meetsThreshold(covered, total, threshold int) bool { + if total == 0 { + return true + } + return covered*100 >= total*threshold } // sh runs an external command with stdio wired through. Every call site From 32ae077f5b52f68be3a2afe15055e49f08abd245 Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Thu, 14 May 2026 15:14:53 -0400 Subject: [PATCH 12/13] fix(discovery): skip onAttempt during ctx cancellation in RetryRefresh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Gemini Code Assist review on PR #125 (commit f854274, finding 1b): when context cancellation races with an in-flight Refresh, the err surfaced from Refresh is a downstream reflection of the shutdown signal — not a real diagnostic. Without a guard, the wavehouse boot path's onAttempt callback writes that error into BootState as "schema discovery: context canceled" which leaks into /health responses during the shutdown window. Functionally harmless but noise an operator could spend cycles trying to debug. Single-line guard at the callback site: } else if ctx.Err() == nil && onAttempt != nil { onAttempt(err) } select-against-ctx.Done() at the loop bottom is unchanged; this only suppresses the callback, not the cancellation-return path. New test TestRetryRefresh_DoesNotFireOnAttemptDuringCancel pins the behaviour with a pre-cancelled ctx + counted-callback fake. Uses the existing fakeConn pattern (Query doesn't inspect ctx, so it returns the staged "transient" error rather than ctx.Canceled — exercising the more interesting case where the error LOOKS real but ctx.Err() reveals shutdown). Pushed back on Gemini's other findings on this commit in their respective threads: 1a) time.After timer leak — Go 1.23+ runtime fix; Gemini already conceded this in round 1 (2026-05-12T22:50:02Z), re-flagged twice since. 1c) Return last underlying error vs ctx.Err() — current design separates BootState (diagnostic) from RetryRefresh return (lifecycle signal); Gemini agreed in round 1. 3) Wait initialBackoff before first retry — trades fast recovery from transient (~ms) for unconditional 2s wait on every boot-degraded case; wrong direction. Filed Gemini's [SHOULD] jitter finding as #141 for clustered-mode follow-up — N=1 single-instance doesn't have thundering-herd risk, adding the knob and the test surface is wasted complexity until clustered mode lands. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/discovery/discovery.go | 10 +++++++- internal/discovery/discovery_test.go | 34 ++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index e5d7478f..9dedcaa0 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -142,7 +142,15 @@ func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxB for { if err := sr.Refresh(ctx); err == nil { return nil - } else if onAttempt != nil { + } else if ctx.Err() == nil && onAttempt != nil { + // Skip the callback when Refresh's error is just a downstream + // reflection of ctx cancellation — that's a shutdown signal, + // not a real diagnostic. Without this guard, a clean shutdown + // fires onAttempt with `context.Canceled`, which the wavehouse + // boot path then writes into BootState as + // "schema discovery: context canceled" + // — visible to anyone curl'ing /health during the shutdown + // window. Not wrong, just noise. onAttempt(err) } select { diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go index 5e9464f5..2cd045bc 100644 --- a/internal/discovery/discovery_test.go +++ b/internal/discovery/discovery_test.go @@ -490,6 +490,40 @@ func TestRetryRefresh_ReturnsOnContextCancel(t *testing.T) { } } +// TestRetryRefresh_DoesNotFireOnAttemptDuringCancel pins the shutdown-noise +// guard added per Gemini's review on f854274: when the context is already +// cancelled going into the loop, the failed Refresh's error is a downstream +// reflection of cancellation (or arrives before the select can catch +// ctx.Done() — either way), and surfacing it via onAttempt would write +// "context canceled" into BootState as a spurious /health diagnostic during +// the normal shutdown window. The guard is a single `ctx.Err() == nil` +// check on the callback site; this test pins it. +func TestRetryRefresh_DoesNotFireOnAttemptDuringCancel(t *testing.T) { + t.Parallel() + // fakeConn doesn't inspect ctx, so the first Refresh against a + // pre-cancelled ctx returns "transient" rather than ctx.Canceled. + // That's exactly the case the guard needs to handle — the error + // looks real, but ctx.Err() reveals we're shutting down anyway. + conn := &fakeConn{errsThenSuccess: []error{errors.New("transient")}} + sr, _ := newFakeRegistry(t, nil) + sr.conn = conn // override the no-error conn from newFakeRegistry + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already cancelled before RetryRefresh starts + + var attempts atomic.Int32 + err := sr.RetryRefresh(ctx, time.Second, time.Second, func(_ error) { + attempts.Add(1) + }) + + require.ErrorIs(t, err, context.Canceled) + assert.Equal(t, int32(0), attempts.Load(), + "onAttempt must not fire when ctx is already cancelled — that would surface shutdown as a /health diagnostic") + // Sanity: Refresh actually ran at least once (so we exercised the + // callback site, not a short-circuit that skipped Refresh entirely). + assert.Equal(t, int32(1), conn.calls.Load(), "Refresh should have been called exactly once before the select caught ctx.Done()") +} + // TestRetryRefresh_BackoffIsBounded verifies that maxBackoff caps the // exponential growth. We use small bounds so the test stays fast. func TestRetryRefresh_BackoffIsBounded(t *testing.T) { From 04709d8c0e0b0a2eca564d0a1a3cab38e40d596a Mon Sep 17 00:00:00 2001 From: Eric Andrechek Date: Mon, 18 May 2026 09:17:41 -0400 Subject: [PATCH 13/13] refactor(main): tie KV/DLQ inits to process-lifetime ctx MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EnsureDLQStream, policy.NewStore, and pipes.NewStore previously took context.Background(). Since #125 creates the process-lifetime ctx earlier (for the schema discovery retry goroutine), these three init calls can now share it — letting SIGINT/SIGTERM cancel them cleanly rather than relying on Docker SIGKILL as the only escape. All three operate against embedded JetStream (in-process), so the practical risk of a hang is near zero — but the inconsistency with the surrounding ctx usage was noted in PR #125 review (taitelee) and the change is mechanical with zero behavioral impact in the happy path. Co-Authored-By: Claude Opus 4.7 (1M context) --- cmd/wavehouse/main.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index dcdc6308..aa975abc 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -236,7 +236,7 @@ func run() int { // DLQ stream. if cfg.DLQ.Enabled { - if err := api.EnsureDLQStream(context.Background(), embeddedMQ.JetStream(), maxBytes/10); err != nil { + if err := api.EnsureDLQStream(ctx, embeddedMQ.JetStream(), maxBytes/10); err != nil { logger.Error("dlq stream init", "error", err) return 1 } @@ -252,14 +252,14 @@ func run() int { defer func() { _ = tiered.Close() }() // Policy store (NATS KV + optional file bootstrap). - policyStore, err := policy.NewStore(context.Background(), embeddedMQ.JetStream(), cfg.Policy.FilePath, logger) + policyStore, err := policy.NewStore(ctx, embeddedMQ.JetStream(), cfg.Policy.FilePath, logger) if err != nil { logger.Error("policy store init", "error", err) return 1 } // Pipes store (NATS KV + optional SQL file directory). - pipesStore, err := pipes.NewStore(context.Background(), embeddedMQ.JetStream(), cfg.Pipes.Dir, logger) + pipesStore, err := pipes.NewStore(ctx, embeddedMQ.JetStream(), cfg.Pipes.Dir, logger) if err != nil { logger.Error("pipes store init", "error", err) return 1