refactor(ingest)!: insert-only pipeline; mutations via /v1/query - #164
Conversation
github-actions
Bot
added
documentation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughMove raw SQL to an admin/service-gated ClickHouse HTTP proxy, add native ClickHouse exec/query dispatch and deterministic query cache-keying, convert Bento ingest to insert-only with DoubleAck-drop on non-insert actions, and update router, policy, SDK, tests, integration wiring, and documentation to match. ChangesAdmin HTTP proxy & tests
Native ClickHouse exec & cache key
Pipes / structured queries
Router, policy, SDK & clients
Ingest worker (Bento) changes
cmd wiring, integration and e2e tests
Docs, changelog, README, security, and checklist
Sequence Diagram (admin proxy flow) sequenceDiagram
participant Client
participant WaveHouseAPI
participant ClickHouse
Client->>WaveHouseAPI: POST /v1/admin/query { "sql": "..." }
WaveHouseAPI->>ClickHouse: HTTP POST (body=raw SQL, X-ClickHouse-User/Key, params=default_format=JSON,date_time_output_format=iso)
ClickHouse-->>WaveHouseAPI: 200 + JSON { "data": [...] } OR 200 + raw body OR non-200 + plain text
WaveHouseAPI->>Client: 200 + application/json (data) OR 200 + [] OR pass-through raw Content-Type OR JSON error (mapped 400/502)
Estimated code review effort 🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related issues
Possibly related PRs
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly refactors the data mutation strategy by enforcing an insert-only ingest pipeline and centralizing all other mutation operations to the Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
taitelee
changed the title
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
|
1 [MUST], 1 [SHOULD], 0 [MAY] — see inline threads for detail. Iterate — fix the mutation caching bug before merge. The The integration test quality note ( |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/api/query.go (1)
74-104:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winMutation SQL is being cached and singleflight-collapsed before execution
Line 74 through Line 104 applies cache lookup/write and
singleflightto all SQL, including mutating statements now routed via/v1/query. That can cause identicalDELETE/UPDATE/TRUNCATE/...requests to return cached[](or share one in-flight result) without executing every call, which is unsafe for write semantics.Suggested fix
@@ if req.SQL == "" { writeJSONError(w, http.StatusBadRequest, "missing sql") return } + isMut := isMutation(req.SQL) cacheKey := queryCacheKey(req.SQL, req.Params) - // Try cache. - if h.Cache != nil { + // Try cache for read-only queries only. + if !isMut && h.Cache != nil { if data, _, err := h.Cache.Get(r.Context(), cacheKey); err == nil && data != nil { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Cache", "HIT") _, _ = w.Write(data) return } } - // Execute query with singleflight to protect ClickHouse from thundering herds. - v, err, _ := h.sf.Do(cacheKey, func() (interface{}, error) { + exec := func() (interface{}, error) { queryCtx, cancel := context.WithTimeout(r.Context(), 30*time.Second) defer cancel() result, err := h.executeQuery(queryCtx, req.SQL, req.Params) if err != nil { return nil, err @@ - if h.Cache != nil { + if !isMut && h.Cache != nil { _ = h.Cache.Set(r.Context(), cacheKey, data, h.DefaultTTL) } return data, nil - }) + } + + var ( + v interface{} + err error + ) + if isMut { + v, err = exec() + } else { + // Read-only queries only: collapse duplicate in-flight work. + v, err, _ = h.sf.Do(cacheKey, exec) + } if err != nil { writeJSONError(w, http.StatusInternalServerError, err.Error()) return } @@ - w.Header().Set("X-Cache", "MISS") + if isMut { + w.Header().Set("X-Cache", "BYPASS") + } else { + w.Header().Set("X-Cache", "MISS") + } _, _ = w.Write(v.([]byte)) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/api/query.go` around lines 74 - 104, The current flow uses queryCacheKey, h.Cache and h.sf.Do for every req.SQL including mutating statements; change it so mutating SQL (e.g., statements starting with INSERT, UPDATE, DELETE, TRUNCATE, ALTER, CREATE, DROP, MERGE, etc.) bypass both the cache and singleflight. Add a small helper (e.g., isMutatingSQL(sql string) bool) and in the handler branch: if isMutatingSQL(req.SQL) then call h.executeQuery with a timed context (like queryCtx used now), json.Marshal the result and write the response directly without calling h.sf.Do, without reading from or writing to h.Cache (and optionally set X-Cache: MISS), otherwise keep the existing singleflight + cache behavior that uses h.sf.Do, queryCacheKey, DefaultTTL and Cache.Set.internal/ingest/bento_test.go (1)
183-213:⚠️ Potential issue | 🟠 Major | ⚡ Quick winTests currently encode legacy “missing action is allowed” behavior
TestJsInput_Read_InsertMessageuses an envelope with noaction, andTestJsInput_Read_NonInsertActionRejectedomits an empty/missing-action case. That leaves the insert-only contract unpinned for the “absent action” path.Please make insert-path tests send explicit
"action":"insert"and add rejection coverage for missing/empty action (DoubleAck + drop + advance-to-next-insert).As per coding guidelines:
internal/ingest/*.go: “The ingest pipeline is insert-only: envelopes with any action other than 'insert' (or absent) are DoubleAck'd and dropped”.Also applies to: 722-748
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/ingest/bento_test.go` around lines 183 - 213, Update the insert-path test to explicitly set "action":"insert" in the EventMessage used by TestJsInput_Read_InsertMessage (modify the evt variable in that test) so the insert contract is pinned, and add/extend a test (e.g., TestJsInput_Read_MissingOrEmptyActionRejected) that sends an envelope with no action or an empty action string and asserts the ingest pipeline rejects it: Read should return a nil message, the returned ack function should DoubleAck the underlying bentoMockMsg (check natsMsg.doubleAcked == true), the message should be dropped (no inFlight increment on jsInput.inFlight), and processing should advance to the next message; reference jsInput.Read, TestJsInput_Read_InsertMessage, and the bentoMockMsg.doubleAcked flag when locating changes.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/src/content/docs/api.md`:
- Line 123: The sentence under the "Insert-only." paragraph is too long; split
the second sentence into two for readability by breaking at the semicolon (after
"`REPLACE`, etc.") and making the second sentence start with "These mutations
must be issued..." while preserving the reference to `POST /v1/query`, the
`admin` / `service` role (or policy role with `RawSQL: true`), and the
explanation that the policy engine authorizes mutations by inspecting written
columns and cannot handle predicate-driven mutations like `DELETE … WHERE`; keep
all technical terms and links intact (e.g., the `POST /v1/query` anchor and
`RawSQL: true`) and ensure punctuation and capitalization are adjusted for two
clear sentences.
In `@internal/ingest/bento.go`:
- Around line 121-122: The current check allows an empty raw.Action as valid;
update the condition so any action other than the literal "insert" — including
the empty string — is rejected and DoubleAck'd. Locate the branch using
raw.Action and slog.WarnContext (the code that logs "rejecting non-insert
message: ingest pipeline is insert-only") and change the if condition to treat
raw.Action == "" the same as other non-"insert" values (e.g., if raw.Action !=
"insert" { ... } or explicitly if raw.Action == "" || raw.Action != "insert" {
... }), then ensure the existing DoubleAck/drop logic is executed for that case.
---
Outside diff comments:
In `@internal/api/query.go`:
- Around line 74-104: The current flow uses queryCacheKey, h.Cache and h.sf.Do
for every req.SQL including mutating statements; change it so mutating SQL
(e.g., statements starting with INSERT, UPDATE, DELETE, TRUNCATE, ALTER, CREATE,
DROP, MERGE, etc.) bypass both the cache and singleflight. Add a small helper
(e.g., isMutatingSQL(sql string) bool) and in the handler branch: if
isMutatingSQL(req.SQL) then call h.executeQuery with a timed context (like
queryCtx used now), json.Marshal the result and write the response directly
without calling h.sf.Do, without reading from or writing to h.Cache (and
optionally set X-Cache: MISS), otherwise keep the existing singleflight + cache
behavior that uses h.sf.Do, queryCacheKey, DefaultTTL and Cache.Set.
In `@internal/ingest/bento_test.go`:
- Around line 183-213: Update the insert-path test to explicitly set
"action":"insert" in the EventMessage used by TestJsInput_Read_InsertMessage
(modify the evt variable in that test) so the insert contract is pinned, and
add/extend a test (e.g., TestJsInput_Read_MissingOrEmptyActionRejected) that
sends an envelope with no action or an empty action string and asserts the
ingest pipeline rejects it: Read should return a nil message, the returned ack
function should DoubleAck the underlying bentoMockMsg (check natsMsg.doubleAcked
== true), the message should be dropped (no inFlight increment on
jsInput.inFlight), and processing should advance to the next message; reference
jsInput.Read, TestJsInput_Read_InsertMessage, and the bentoMockMsg.doubleAcked
flag when locating changes.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 17e11d35-e61a-4283-b7e0-62009163f6a2
📒 Files selected for processing (13)
AGENTS.mdCHANGELOG.mdcmd/wavehouse/main.godocs/src/content/docs/api.mddocs/src/content/docs/architecture.mdinternal/api/query.gointernal/api/query_test.gointernal/ingest/bento.gointernal/ingest/bento_test.gotests/integration/dlq_test.gotests/integration/ingest_test.gotests/integration/query_test.gotests/integration/setup_test.go
💤 Files with no reviewable changes (3)
- cmd/wavehouse/main.go
- tests/integration/dlq_test.go
- tests/integration/setup_test.go
github-project-automation
Bot
moved this from Backlog
to In review
in WaveHouse Task Board
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request transitions the ingest pipeline to an insert-only model, removing inline delete handling and requiring all non-insert mutations (DELETE, UPDATE, TRUNCATE, etc.) to be performed via the /v1/query endpoint. The query handler is updated to classify SQL verbs and route mutations through Exec to return an empty JSON array instead of an error. Feedback identifies a critical caching bug where mutation results could be incorrectly persisted in the API layer, suggests extending SQL comment stripping to support the # prefix used in ClickHouse, and notes a minor typo in the documentation comments. Iterate.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/api/query_test.go`:
- Around line 171-217: Add table-driven test rows in TestIsMutation to cover the
remaining mutation verbs ATTACH, DETACH, KILL, SET, and USE: for each verb add
at least one case (e.g., "ATTACH TABLE t", "DETACH TABLE t", "KILL QUERY 1",
"SET x = 1", "USE db") and assert they return true via isMutation; place them
alongside the other true cases so the test exercises leading whitespace/comments
behavior similarly. Ensure the test names are unique (e.g., "attach", "detach",
"kill", "set", "use") and follow the existing t.Run/t.Parallel pattern.
In `@tests/integration/ingest_test.go`:
- Around line 86-87: The test's context timeout created via context.WithTimeout
(ctx, 60*time.Second) is too short for two sequential require.Eventually blocks
plus HTTP/JetStream/DB overhead; update the timeout duration to a larger value
(e.g., 120s or 180s) by changing the call to context.WithTimeout where ctx and
cancel are defined so the test has sufficient headroom for both Eventually
blocks and external delays.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 8dd924a8-767c-46d3-8874-f1cd88a438ee
📒 Files selected for processing (9)
AGENTS.mdCHANGELOG.mddocs/src/content/docs/api.mddocs/src/content/docs/architecture.mdinternal/api/query.gointernal/api/query_test.gointernal/ingest/bento.gointernal/ingest/bento_test.gotests/integration/ingest_test.go
2 participants
Summary
Three-part lockdown of the data path on the way to the alpha release. Each part landed as its own commit on this branch.
1. Lock the Bento ingest pipeline to inserts only. The
action: "delete"branch injsInput.Readis gone; any envelope whoseactionisn't"insert"(or absent) isDoubleAck'd and dropped. Thedelete-envelopeDLQ shape and theWave-DLQ-Typeheader it required are gone too —dlq.<table>now carries only insert-failure payloads, so DLQ consumers no longer need to discriminate.2. Move raw-SQL passthrough to
POST /v1/admin/queryunder admin gating. Previously/v1/querywas registered under/v1's auth middleware with the authorization decision deferred to the handler — any caller (or, withauth.enabled=false, anyone) could submit raw SQL, and apolicy.RolePermissions.raw_sql: truegrant on any table let non-admins through. The route now lives under/v1/admin/*, gated by the sameRequireRole("admin","service")middleware that covers the rest of the admin tree. Thepolicy.RolePermissions.raw_sqlfield is removed outright — equivalent capability is "issue a JWT with role=admin or role=service."3. Replace the handler's
clickhouse-goQuery/Exec dispatch with a thin HTTP proxy. Instead of classifying the leading SQL verb client-side (~150 LOC ofisMutation/ CTE parser / comment stripper) and routing betweendriver.Execanddriver.Query, the handler now POSTs the SQL verbatim to ClickHouse's HTTP interface and forwards the response. Wins: multi-statement input works (SELECT 1; TRUNCATE t), any DDL/DML/SYSTEM verb (current or future) works without WaveHouse code changes, ClickHouse's own error messages reach the admin verbatim, no cache, no singleflight,Cache-Control: no-storeon every response. The verb-classification primitives still exist (now ininternal/api/clickhouse_exec.go) because the structured-query and pipes handlers still need them —clickhouse-go's nativeQuery()errors on no-result-set statements, so explicit Exec-vs-Query routing remains correctness for those paths.Why this shape: the policy engine authorizes mutations by inspecting the payload of an operation, which works for inserts but doesn't extend to predicate-driven mutations — we can't prove a
WHEREclause is satisfiable only for rows the caller can touch, andWHERE 1=1would otherwise nuke a table. Rather than ship a partial enforcement story, the pipelined surface narrows to what policy can authorize today, and the escape hatch for everything else lives behind admin-equivalent role gating with no per-statement scope check (the role gate is the entire authorization story; admins/service tokens see exactly what ClickHouse sees).Paired fix (closes #118): the original
/v1/queryreturned HTTP 500 forTRUNCATE/DROP/DELETE/etc. because clickhouse-go'sdriver.Query()errors on no-result-set statements. The verb-classification fix lived in this branch before the HTTP-proxy rewrite; the proxy now sidesteps the issue entirely (ClickHouse decides what to do natively), but the structured-query/pipes paths still rely on the same primitives viaexecuteCHQuery.Breaking surface
action: "delete"envelopes oningest.<table>are no longer honored.StartIngestWorkersignature drops itschConn driver.Connparameter (andjsInput.chConn/jsInput.jsfields go with it).Wave-DLQ-Type: delete-envelopeNATS header is gone. DLQ consumers that branched on it should drop the branch.POST /v1/query→POST /v1/admin/query. The old URL returns 404.policy.RolePermissions.raw_sqlfield removed from the policy schema. Operators withraw_sql: trueinpolicy.yamlwill see a YAML-load warning for the unknown field but it's otherwise ignored — the capability is now expressed solely via JWT role./v1/admin/queryno longer accepts aparamsarray. ClickHouse's HTTP interface uses named{name:Type}binding, not positional?. Inline literals or use the structured query endpoint (POST /v1/tables/{table}/query) for safe binding from user inputs.client.sql(sql, params)drops theparamsargument.api.NewQueryHandlersignature is(endpoint, username, password, database string), replacing(driver.Conn).Test plan
TestJsInput_Read_NonInsertActionRejected: table-driven across non-insert verbs; asserts DoubleAck (never Nak) and thatReadadvances.TestNewRouter_RawSQLAdminGate: admin reaches handler, service reaches handler, viewer → 403, no-role+auth-off → reaches handler, no-role+auth-on → 401.TestQueryHandler_*(proxy contract): SQL forwarded verbatim with the correct query-string params and headers,{meta,data,rows}envelope extracted to bare data array, empty-body mutation →[], ClickHouse error forwarded with its own message,Cache-Control: no-storeset, no auth headers when blank, raw forward on unexpected non-JSON 200 body, request context cancellation propagates to upstream.TestExecuteCHQuery_MutationRoutesToExec/SelectRoutesToQuery(inclickhouse_exec_test.go): keeps the still-required structured-query + pipes Exec/Query dispatch covered.TestIngest_NonInsertActionDropped: publishes a delete envelope to JetStream after a successful insert; asserts the row stays and nothing leaks to the DLQ.TestQuery_TruncateReturnsEmptyArray: pins bug:TRUNCATEvia/v1/queryreturns HTTP 500 #118 end-to-end through the proxy now (HTTP 200 +[], row actually disappears).TestQuery_DeleteReturnsEmptyArray: predicate-drivenDELETEthrough/v1/admin/query; targeted row goes, sibling row stays. Updated to inline the literal since positional?is gone.client.sql()posts to/v1/admin/queryand never sends aparamsfield.make verify,make test-unit,make test-sdk,go build -tags=integration ./...,go vet -tags=integration ./...all green locally before push.Related Issues
TRUNCATEvia/v1/queryreturns HTTP 500 #118/v1/admin/queryno longer caches, singleflights, or bypasses cache for mutations. Every request is a fresh round-trip to ClickHouse and the response carriesCache-Control: no-store. Goes further than the issue's acceptance criteria asked./v1/admin/*with theRequireRole("admin","service")gate; newTestNewRouter_RawSQLAdminGatepins the contract). NOT closing feat(auth): RequireRoles middleware — fail closed, no permissive fallback #145 because three items from its Proposed Solution remain open: (a) admin gate on/v1/dlq/stats(still mounted outside/v1/admin/*— any authenticated caller can hit it today); (b) structured WARN log on every role denial withreason/role_observed/roles_allowed/routefields; (c) cosmetic renameRequireRole→RequireRolesper the issue's stated name. Leaving feat(auth): RequireRoles middleware — fail closed, no permissive fallback #145 open so those don't silently drop on merge.servicefrom automatic admin #168 — decoupleservicefrom being a hardcoded admin equivalent, add capability-based per-feature permissions (pipes.read,pipes.write,policy.write,raw_sql,log_level, …).Summary by CodeRabbit
Breaking Changes
raw_sqlremoved.New Behavior
Tests & Docs