fix(query): break streaming loop on Flush error so DuckDB result buffers free on client disconnect

[xe-nvdk]

Summary

• Heavy queries against `/api/v1/query` (and `/api/v1/query/arrow`) hold DuckDB result-set memory until query completion or the 5-min timeout when the client disconnects mid-stream — Grafana panel close, browser tab kill, dashboard refresh. The streaming goroutine has no way to learn the client closed the TCP connection because fasthttp's `RequestCtx.Done()` only fires on server shutdown (`server.go:2719-2745`), and `c.UserContext()` is already cancelled by the time the async stream writer runs.
• The reliable signal in fasthttp's streaming model is the error returned by `bufio.Writer.Flush()` — when the underlying TCP connection is closed, the next flush surfaces EPIPE. The fix captures that error and breaks the streaming loop.
• Same pattern applied to the three streaming write paths: Arrow IPC (`query_arrow.go`), Arrow-to-JSON (`query_arrow_json.go` — what the default `/query` endpoint uses on `-tags=duckdb_arrow` builds), and the pure database/sql JSON fallback (`query_json_writer.go`).
• Aligned the four "stream truncated after headers committed" log sites to `Warn` (from `Error`). Headers are already committed when streamErr fires, so the client got a partial result — by definition the server worked. Client disconnect is expected ops noise.

Origin

Filed in response to a user-reported pattern: heavy time-bucket aggregations and wide GROUP BYs cause RSS to climb during query execution. Investigation found multiple potential leak sources; this PR fixes the one with the clearest causal mechanism and reproducer (client-disconnect path).

A throttled `DuckDB.ClearHTTPCache()` from the query path (parquet_metadata_cache eviction) was investigated and tested in the same branch but dropped from this PR — at 1200 queries on a small local-backend dataset, the with-fix and without-fix runs were indistinguishable (both converged to ~180 MB RSS post-settle). Without empirical evidence at the test scale, shipping it would be cargo-cult. May revisit at production scale.

Test plan

• `go build ./internal/api/... ./cmd/arc/...` clean on both default and `-tags=duckdb_arrow` build configurations.
• `go test -tags=duckdb_arrow ./internal/api/...` — all existing tests pass plus new `TestStreamArrowJSON_FlushErrorBreaksLoop` (7 tests in the Arrow JSON test file).
• New regression test uses `errAfterNBytes` (an `io.Writer` that fails after N bytes) and asserts:
  • the returned error wraps the sentinel via `errors.Is`
  • the row count is bounded to within `jsonFlushInterval + batchRows` of the failure point (i.e. break fires at the next 5000-row Flush boundary, not later)
• Spot-check on Grafana: open a heavy panel against the `/query` endpoint, kill the browser tab mid-load. Pre-fix: DuckDB result reader keeps draining for up to 5 min. Post-fix: reader stops within the next flush window.

Internal review

Four parallel reviewer agents (correctness, security, code-quality, performance) per the CLAUDE.md review protocol. Findings addressed:

• Same bug in non-Arrow JSON path (code-quality HIGH) → applied identical fix in `query_json_writer.go:169`.
• Asymmetric severity Error vs Warn (3 agents) → aligned all four streaming-truncation log sites to Warn.
• Stale test comment + loose assertion (multiple agents) → fixed comment, tightened bound to `jsonFlushInterval + batchRows` instead of `>= totalRows`.
• No IPC-path test coverage (code-quality MEDIUM) → noted as a follow-up. The IPC path uses structurally identical code to the JSON path that the test exercises. Extracting a testable helper from the Fiber handler is a separate refactor.

Pushed back on:

• Convert reader.Release() to defer (correctness MEDIUM) — pre-existing pattern, not introduced by this fix, separate hardening.
• Short-circuit envelope-close on streamErr (correctness HIGH) — the design is intentional ("emit valid JSON regardless" for clients parsing as bytes arrive); cost on disconnect is one no-op JSON marshal.
• Cancel ctx immediately on flush-error break (performance MEDIUM) — re-read the code, this is already correct: the cancel() at end of closure fires after the break, and `reader.Next()` isn't called again because we broke the loop.
• Add sentinel error / metric (multiple LOW) — separate observability PR.
• Convert test to CheckedAllocator (perf NIT) — the existing `buildArrowBatch` helper leaks builder allocations on every test (pre-existing bug); fixing it would scope-creep. Filed mentally as a follow-up.

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request implements logic to abort streaming query execution when a client disconnects by capturing errors from bufio.Writer.Flush(). This prevents unnecessary memory pressure and processing by breaking the streaming loops early when the underlying TCP connection is closed. Additionally, several log entries for truncated streams were downgraded from Error to Warn to reduce operational noise. The feedback suggests further refining this noise reduction by handling potential errors during ipcWriter.Close() and tightening the row count assertion in the new regression test to ensure the loop breaks exactly at the expected boundary.

[xe-nvdk]

@gemini-code-assist round 1 — both taken in 4109b43.

Taken: `ipcWriter.Close()` log downgrade (internal/api/query_arrow.go:201) — you're right. When the loop broke because the client disconnected, `ipcWriter.Close()` is guaranteed to fail flushing trailing IPC metadata over the already-closed connection. Logging that at `Error` defeated the ops-noise reduction. Now `Warn` with a comment explaining the cause/effect.

Taken: tighter test assertion (internal/api/query_arrow_json_test.go) — I added a temporary `t.Logf` to verify the actual rowCount before tightening:

```
query_arrow_json_test.go:415: actual rowCount=5000, jsonFlushInterval=5000
```

Empirically: bufio's internal 4 KiB auto-flush hits the failing writer well before row 5000 and stores a sticky error; subsequent `WriteByte`/`WriteString` calls silently no-op until the explicit `Flush()` at row 5000 returns the stored error. So the exact-boundary check holds. Changed `rowCount > tighterBound` to `rowCount != jsonFlushInterval`.

Ready for another pass.

[gemini-code-assist]

Code Review

This pull request introduces mechanisms to abort streaming query results upon client disconnection by monitoring flush errors, thereby preventing unnecessary memory consumption from draining result sets. It also addresses S3 endpoint normalization for DuckDB and fixes DELETE operations on non-TLS S3 storage. Review feedback suggests that the 5000-row interval for error checking may still allow significant wasted processing and recommends more frequent checks. Furthermore, the feedback notes that downgrading all stream-truncation logs to a warning level might obscure genuine server-side errors and points out inconsistencies in error reporting and metrics across different streaming handlers.

[xe-nvdk]

@gemini-code-assist round 2 — all four findings addressed in 743736b. The change ended up bigger than expected because finding #2 required introducing the sentinel error to disambiguate cleanly.

HIGH 1+2: wasted work between flush boundaries (query_arrow_json.go:287, query_json_writer.go:176)

Considered per-row Write-error check but rejected as too expensive for hot-path queries — adds a branch per WriteByte across millions of rows, and `bufio.Writer` has no public API to inspect its stored error without forcing a Flush (which incurs a syscall when buffered bytes exist).

Lowered `jsonFlushInterval` from 5000 to 1000. Compromise: 5× faster disconnect detection, ~5× more no-op Flush calls (essentially free), worst-case wasted formatting work bounded to ~1000 rows (~10-100µs at typical query rates). Acceptable.

MEDIUM 1: log level disambiguation (query.go:1536)

You're right — downgrading every stream-truncation error to Warn hid genuine server-side failures (scanner errors, db iteration failures). Fixed by introducing `errClientDisconnected` sentinel (query.go:107-114) and wrapping the Flush-failure sites with it. Caller log sites now disambiguate:

```go
ev := h.logger.Error()
if errors.Is(streamErr, errClientDisconnected) ||
errors.Is(streamErr, context.DeadlineExceeded) ||
errors.Is(streamErr, context.Canceled) {
ev = h.logger.Warn()
}
```

Applied at four call sites: `query_arrow_json.go`, `query.go` (×3 — two streamTypedJSON callers and queryMeasurement), and the newly-refactored `query_arrow.go` (see MEDIUM 2 below).

MEDIUM 2: Arrow IPC handler inconsistency (query_arrow.go:201)

Caught a real asymmetry. The IPC handler logged inline and broke without setting a streamErr, so the closure always logged "Arrow streaming query completed" even on disconnect, AND `IncQueryErrors` was never incremented for those events. Refactored to mirror the JSON handler: capture all error conditions (ctx cancel, decimal cast, ipcWriter.Write, Flush) into `streamErr`, log/metric once at closure end with the same severity disambiguation.

Test update: added an `errors.Is(err, errClientDisconnected)` assertion alongside the existing underlying-I/O-sentinel check. Tests confirm both ancestors are in the error chain.

Also closes #428 (sentinel was deferred there as "defer until a caller actually needs to branch" — gemini's log-disambiguation finding made the need concrete).

Ready for another pass.

[gemini-code-assist]

Code Review

This pull request introduces a mechanism to abort streaming queries when a client disconnects by monitoring errors from bufio.Writer.Flush(). This prevents unnecessary resource consumption and memory usage for abandoned requests. The logging logic has been updated to categorize client-side disconnects and timeouts as warnings rather than errors. The PR also includes fixes for S3 endpoint scheme normalization and DELETE operations on non-TLS S3. Review feedback recommends extracting the repeated error-checking logic into a shared helper function and updating a stale comment in the test suite that references the old flush interval.

[xe-nvdk]

@gemini-code-assist round 3 — both findings taken in aa70be1.

Taken: `isClientError` helper (internal/api/query.go:116) — you're right, the 4-line errors.Is block was repeated 5 times. Extracted two helpers:

```go
func isClientError(err error) bool {
return errors.Is(err, errClientDisconnected) ||
errors.Is(err, context.DeadlineExceeded) ||
errors.Is(err, context.Canceled)
}

func (h *QueryHandler) streamErrEvent(err error) *zerolog.Event {
if isClientError(err) {
return h.logger.Warn()
}
return h.logger.Error()
}
```

Each of the 5 sites now reads as a single line:

```go
h.streamErrEvent(streamErr).Err(streamErr).
Int("rows_sent", rowCount).
Float64("execution_time_ms", ...).
Msg(...)
```

Net -11 lines across the diff. Behaviour is identical (same severity disambiguation, same call shape), just expressed once instead of five times.

Taken: stale test comment — `jsonFlushInterval (5000)` → references the constant symbolically now so it doesn't drift again if we ever retune the interval.

Ready for another pass.

[gemini-code-assist]

Code Review

This pull request introduces a mechanism to detect client disconnects during streaming queries by monitoring bufio.Writer.Flush() errors, preventing unnecessary processing and memory usage for abandoned connections. It also includes logic to differentiate between client-side operational noise (logged as warnings) and server-side failures (logged as errors). Other changes include S3 endpoint normalization for DuckDB and a fix for DELETE operations on non-TLS S3. Review feedback suggests standardizing error messages across the various streaming handlers for better consistency.