Enable telemetry by default with DSN-controlled priority and full event correctness tests

[samikshya-db]

Summary

This PR extends the telemetry implementation across two areas.

1. DSN / config changes (original scope)

• EnableTelemetry *bool tristate in telemetry.Config: nil = defer to server flag, &true = client opt-in, &false = client opt-out.
• Two-level enable priority: DSN enableTelemetry=true → always on; otherwise use server feature flag.
• Two new DSN params: telemetry_retry_count and telemetry_retry_delay.

2. Telemetry gap fixes (new in this PR)

Four correctness bugs found during end-to-end testing against a real warehouse:

EXECUTE_STATEMENT / CLOSE_STATEMENT silently lost on shutdown
Root cause: agg.cancel() fired while a worker was mid-HTTP-export.
Fix: added inFlight sync.WaitGroup; close() calls inFlight.Wait() before cancel().

total_chunks_present: null for paginated CloudFetch
Root cause: server reports 1 link per FetchResults call; grand total never in a single response.
Fix: pass r.chunkCount through closeCallback; connection.go sets chunk_total_present if the server never reported it.

operation_latency_ms: null for CLOSE_STATEMENT
Root cause: CloseOperation RPC completes in <1ms → rounds to 0; omitempty drops 0.
Fix: removed omitempty from OperationLatencyMs.

CloudFetch S3 timing fields not populated
Root cause: per-S3-file download time was not measured.
Fix: added onFileDownloaded func(downloadMs int64) callback to cloudIPCStreamIterator; connection.go aggregates initial/slowest/sum timings.

3. DSN parameters (full set)

Parameter	Type	Default	Description
enableTelemetry	bool	unset	Overrides server flag when set
telemetry_batch_size	int	100	Events per batch
telemetry_flush_interval	duration	5s	Periodic flush interval
telemetry_retry_count	int	3	Max retry attempts on export failure
telemetry_retry_delay	duration	100ms	Base delay between retries (exponential backoff)

Key files changed

• telemetry/aggregator.go — inFlight WaitGroup; 5-step close() ordering
• telemetry/interceptor.go — RecordOperation takes statementID so CLOSE_STATEMENT carries sql_statement_id
• telemetry/request.go — removed omitempty from OperationLatencyMs
• connection.go — closeCallback(latencyMs, chunkCount, err) + cloudFetchCallback wiring
• internal/rows/rows.go — closeCallback passes r.chunkCount; cloudFetchCallback threaded through
• internal/rows/arrowbased/batchloader.go — onFileDownloaded callback per S3 file download

Test plan

• go build ./... — clean compile
• go test ./telemetry/... -count=1 — all pass
• go test ./internal/rows/... -count=1 — all pass
• go test ./... -short -count=1 — full suite passes

New correctness tests

telemetry/aggregator_test.go (new file, 5 tests):

• WaitsForInFlightWorkerExports — close() blocks until all HTTP exports finish, even if workers picked up jobs before the drain step ran
• DrainsPendingQueueJobsBeforeCancel — jobs sitting in exportQueue are exported synchronously during drain
• InFlightAddBeforeSend — inFlight.Add(1) precedes the channel send so no job is invisible to Wait()
• SafeToCallMultipleTimes — concurrent close() calls do not deadlock (sync.Once)
• DropWhenQueueFull — drop path calls inFlight.Done() so Wait() is never permanently blocked

telemetry/integration_test.go (2 new tests):

• OperationLatencyMs_ZeroNotOmitted — raw JSON contains "operation_latency_ms":0, not absent
• ChunkTotalPresent_DerivedFromChunkCount — chunk_total_present tag propagates to ChunkDetails

internal/rows/arrowbased/batchloader_test.go (2 new tests):

• OnFileDownloaded callback invoked once per file with positive downloadMs
• Nil callback is safe on non-telemetry paths (no panic)

internal/rows/rows_test.go (2 new tests):

• CloseCallback_ReceivesChunkCount — callback gets correct total pages after multi-page iteration
• CloseCallback_NilDoesNotPanic — nil closeCallback is safe on rows.Close()

This pull request was AI-assisted by Isaac.

[vikrantpuppala]

[F1] inFlight.Wait() has no timeout — can block conn.Close() indefinitely (Severity: High)

agg.inFlight.Wait() is context-unaware. If a telemetry export hangs (slow server, no HTTP client timeout), close() blocks forever → conn.Close() blocks forever → sql.DB.Close() can hang.

The parent telemetryClient.close() creates a 5s timeout context, but sync.WaitGroup.Wait() doesn't respect context cancellation. The per-connection Interceptor.Close() → flushSync() path has no timeout at all.

Suggested fix: Replace bare Wait() with a select on a channel + context deadline:

waitCh := make(chan struct{})
go func() { agg.inFlight.Wait(); close(waitCh) }()
select {
case <-waitCh:
case <-ctx.Done():
    logger.Debug().Msg("telemetry: close timed out waiting for in-flight exports")
}

Also pass a timeout context from Interceptor.Close().

(Flagged by 3 reviewers: ops, performance, security)

---

Code Review Squad — feedback

[vikrantpuppala]

[F2] retryCount=0 silently ignored — user cannot disable retries via DSN (Severity: High)

DSN parsing at config.go:256 accepts telemetry_retry_count=0 (retryCount >= 0). The test TestParseTelemetryConfig_RetryCountZero asserts MaxRetries=0 means "disable retries." But here, the guard retryCount > 0 silently discards 0, defaulting to 3 retries.

A user setting telemetry_retry_count=0 in their DSN expects no retries but gets 3.

Suggested fix: Change to if retryCount >= 0 to match the DSN parser and test intent. Or use a sentinel (-1) for "not set."

(Flagged by 4 reviewers: agent-compat, language, architecture, devils-advocate)

---

Code Review Squad — feedback

[vikrantpuppala]

[F3] inFlight.Done() not deferred — panic in export() causes permanent deadlock (Severity: High)

If exporter.export() panics (edge case in JSON marshal or HTTP transport), inFlight.Done() is never called, and inFlight.Wait() in close() blocks forever. While export() has internal panic recovery, defense-in-depth suggests protecting the WaitGroup balance.

Suggested fix:

case job, ok := <-agg.exportQueue:
    if !ok { return }
    func() {
        defer agg.inFlight.Done()
        agg.exporter.export(job.ctx, job.metrics)
    }()

(Flagged by: devils-advocate)

---

Code Review Squad — feedback

[vikrantpuppala]

[F4] Callback parameter proliferation: 8-param NewRows, 5-param telemetryUpdate (Severity: Medium)

Three function-typed params with anonymous signatures, two nullable. All non-telemetry callers pass nil, nil, nil. The telemetryUpdate callback has 5 positional numeric params where 0 is a sentinel value.

Suggested fix: Bundle into a struct:

type RowsTelemetryCallbacks struct {
    OnChunkFetched   func(data ChunkFetchEvent)
    OnClose          func(latencyMs int64, chunkCount int, err error)
    OnCloudFetchFile func(downloadMs int64)
}

Pass *RowsTelemetryCallbacks (nil = no telemetry) instead of three separate function params.

(Flagged by 4 reviewers: agent-compat, language, maintainability, architecture)

---

Code Review Squad — feedback

[vikrantpuppala]

[F8] Shared mutable closure state needs single-goroutine invariant documented (Severity: Medium)

The chunkTimingInitialMs, chunkTimingSlowestMs, chunkTimingSumMs, chunkTimingInitialSet, and cloudFetchFileCount variables are captured by both telemetryUpdate and cloudFetchCallback closures without synchronization.

After verification: both callbacks are invoked from the same goroutine (the caller of rows.Next()) — telemetryUpdate from fetchResultPage and cloudFetchCallback from cloudIPCStreamIterator.Next(). No data race in current code. However, this invariant is undocumented and any future refactoring that adds concurrency would silently introduce a race.

Suggested fix: Add a comment near the variable declarations:

// These variables are safe without a mutex because they are only mutated
// from callbacks invoked sequentially by the single goroutine that calls
// rows.Next() (telemetryUpdate from fetchResultPage, cloudFetchCallback
// from cloudIPCStreamIterator.Next).

(Flagged by 8 reviewers: all — consensus that it's safe today but fragile)

---

Code Review Squad — feedback

[vikrantpuppala]

Code Review Squad Report

Merge Safety Score: 62/100 — HIGH RISK

9 specialized reviewers analyzed this PR in parallel. Below is the aggregated report.

Executive Summary

The telemetry correctness fixes (inFlight WaitGroup, FinalizeLatency, omitempty, CloudFetch timing) are well-motivated and thoroughly tested. Two bugs and a deadlock risk need addressing before merge.

Must Address (High)

ID	Finding	Reviewers
F1	inFlight.Wait() has no timeout — can block conn.Close() indefinitely	ops, performance, security
F2	retryCount=0 silently ignored via DSN — user cannot disable retries	agent-compat, language, architecture, devils-advocate
F3	inFlight.Done() not deferred in exportWorker — panic causes deadlock	devils-advocate

Should Address (Medium)

ID	Finding	Reviewers
F4	Callback parameter proliferation (8-param NewRows, 5-param callback)	agent-compat, language, maintainability, architecture
F5	Dual tristate: *bool vs ConfigValue[bool] for EnableTelemetry	language, maintainability, architecture
F6	Duplicated chunk timing accumulation in telemetryUpdate/cloudFetchCallback	maintainability, performance
F7	OperationLatencyMs omitempty removal affects all metric types (0 vs "not measured")	devils-advocate
F8	Shared mutable closure state needs thread-safety invariant documented	8/9 reviewers

Consider (Low)

ID	Finding
F9	telemetry_retry_delay parse errors silently swallowed (inconsistent with retry_count)
F10	Tag constants defined in tags.go but raw strings used in connection.go and request.go
F11	closeCallback always passes nil/false to AfterExecute/CompleteStatement
F12	goto drained style — use labeled break instead

What's Good

• Telemetry failure isolation is well-designed — all paths have panic recovery, errors swallowed, no query impact
• inFlight WaitGroup shutdown fix correctly prevents lost EXECUTE_STATEMENT/CLOSE_STATEMENT on close
• FinalizeLatency cleanly separates execute-phase timing from row-iteration timing
• *bool tristate correctly models nil/true/false semantics and server flag fallback
• Test coverage is thorough: aggregator close ordering, callback nil-safety, chunk propagation, event correctness

Score Breakdown

Category	Count	Deductions
High	3	-30
Medium	5	-15
Low	4	-4
Total		62/100

---

Code Review Squad — feedback