Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
- **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.
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

### 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.
Expand Down
41 changes: 33 additions & 8 deletions cmd/wavehouse/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
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) {
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
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)
}()
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
}

// State directories — one configurable root, fixed subdir convention.
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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

Expand All @@ -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),
Expand Down
16 changes: 13 additions & 3 deletions docs/api.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved

---

### `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):**

Expand Down
4 changes: 2 additions & 2 deletions docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.

Expand Down
14 changes: 12 additions & 2 deletions docs/deployment.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
- 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:
Expand Down
56 changes: 56 additions & 0 deletions internal/api/health.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -19,13 +60,28 @@ 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)
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
_ = json.NewEncoder(w).Encode(map[string]string{"status": "degraded", "error": err.Error()})
return
}
}
_ = json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
}

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)
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
_ = 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)
Expand Down
84 changes: 84 additions & 0 deletions internal/api/health_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
35 changes: 35 additions & 0 deletions internal/discovery/discovery.go
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
}
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
backoff *= 2
Comment thread
EricAndrechek marked this conversation as resolved.
Resolved
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) {
Expand Down
Loading
Loading