perf: background write batcher for DuckDB telemetry ingest

CLAUDE.md

@@ -255,12 +255,20 @@ POSTGRES_SSLMODE=disable
 # DuckDB telemetry backend (only with -tags telemetry_duckdb build; see "DuckDB Telemetry Backend" below)
 DUCKDB_MEMORY_LIMIT=                  # e.g. 4GB. Unset = DuckDB auto-tunes (~80% RAM). Set explicitly in memory-capped containers to avoid OOM.
 DUCKDB_THREADS=                       # e.g. 4. Unset = DuckDB auto-tunes (= cores). Cap in constrained/shared environments.
-DUCKDB_CHECKPOINT_THRESHOLD=          # e.g. 256MB. Unset = DuckDB default (16MB). Raise under sustained ingest to reduce WAL checkpoint stalls; costs a larger WAL and longer restart replay.
+DUCKDB_CHECKPOINT_THRESHOLD=          # WAL checkpoint threshold. Unset = 256MB (backend default; DuckDB's own 16MB default stalls sustained ingest). Lower it if restart replay time matters more than ingest throughput.
+DUCKDB_WRITE_QUEUE_ROWS=131072        # background write batcher: per-table queue capacity in rows. A full queue answers 503 + Retry-After (load shedding before the ack).
+DUCKDB_WRITE_FLUSH_ROWS=32768         # flush the per-table appender after this many rows
+DUCKDB_WRITE_FLUSH_INTERVAL_MS=100    # ...or after this age, whichever first. Bounds read-lag and the acked-data loss window on crash.
+DUCKDB_WRITE_WRITERS=                 # writer goroutines per hot table. Unset = 2 x CPU cores, capped at 8 (VictoriaMetrics-style drain concurrency).
+DUCKDB_WRITE_QUEUE_WAIT_MS=2000       # how long enqueue waits for queue space before the 503 (VictoriaMetrics-style insert queueing). Bursts the writers sustain on average are absorbed instead of rejected.
 
 # Ingest admission gate (all telemetry ingest endpoints: /api/report, /api/profiles/ingest, /api/otel/*)
 INGEST_MAX_CONCURRENT=                # max concurrently processed ingest requests. Unset = 2×CPU cores, min 4. Bounds ingest memory so overload sheds load with 503s instead of the process being OOM-killed (on DuckDB an OOM death is followed by a minutes-long WAL-replay stall on restart).
 INGEST_ADMISSION_WAIT_SECONDS=5       # how long a request may wait for a slot before the 503 + Retry-After; 0 = reject immediately when saturated
 
+# Profiling (opt-in)
+PPROF_PORT=                           # e.g. 6060. Serves net/http/pprof on 127.0.0.1:<port> (localhost only; unreachable through the public router). Unset = disabled.
+
 # Notifications
 NOTIFICATION_POLL_SECONDS=60          # polled rule evaluation interval; minimum 5, invalid values fall back to 60
 
@@ -799,8 +807,8 @@ for _, item := range items {
 Built with `-tags telemetry_duckdb` (`CGO_ENABLED=1` required), this is an alternative telemetry store for the same `DB_TYPE=sqlite` deployment: the **main DB stays SQLite** (`db.DB`, relational/config), while the **telemetry DB becomes DuckDB** (`db.TelemetryDB`, columnar). It exists because DuckDB's columnar engine is dramatically faster on the analytics/aggregation reads the dashboard issues — at 10M rows it clears read-probe thresholds that SQLite times out on. Backends are selected on two build-tag axes: `telemetry_ch` / `telemetry_duckdb` / *(none = SQLite telemetry)* for the telemetry store and `transactional_pg` / *(none = SQLite main)* for the relational store. Only three combinations are supported — *(no tags)* dual SQLite, `telemetry_duckdb`, and `transactional_pg telemetry_ch` — enforced by compile-time guard files in `backend/app/db/` (stale `pgch`/`duckdb`/`oltp_*` tags also fail with a rename message). Repositories are organized on the same two axes: telemetry repositories live in per-backend packages `backend/app/repositories/telemetry/{clickhouse,sqlite,duckdb}/` and transactional (relational) repositories in `backend/app/repositories/transactional/{pg,sqlite}/`, each re-exported as singletons through tag-guarded facade files at the axis package root (`telemetry/telemetry_ch.go` etc., `transactional/transactional_pg.go` etc.). Consumers import the facade packages — `telemetry.SpanRepository`, `transactional.UserRepository` — never a backend package directly. Helpers shared by all telemetry backends are in `telemetry/shared/`, the SQLite scan/value types shared by the sqlite+duckdb backends are in `telemetry/sqlitetypes/`, and helpers/types shared by the transactional backends (auth-token hashing/time formats, facade-crossing structs) are in `transactional/shared/`. The `transactional/pg` and `transactional/sqlite` implementations are intentionally kept dialect-neutral (lit `:name` queries rendered per `db.Driver`), enforced byte-for-byte by `transactional/parity_test.go`. Running Postgres requires the `transactional_pg` build: the default build's migration runner applies SQLite-dialect migrations unconditionally, so `DB_TYPE=postgres` without the tag is not a supported combination.
 
 - **Driver:** `github.com/duckdb/duckdb-go/v2` (the official driver; marcboeker/go-duckdb is deprecated). Bundles prebuilt static libs for glibc only — **not musl/Alpine**, so the image uses Debian (`Dockerfile.duckdb`).
-- **Opened in** `backend/app/db/db_telemetry_duckdb.go`: telemetry path is the SQLite path with `.db` swapped for `_telemetry.duckdb`. By default DuckDB auto-tunes to the host; `DUCKDB_MEMORY_LIMIT`/`DUCKDB_THREADS`/`DUCKDB_CHECKPOINT_THRESHOLD` (passed through as DSN config options) let operators cap memory/threads so a memory-capped container doesn't read the host's RAM and OOM-kill the backend, and raise the WAL checkpoint threshold (default 16MB) so sustained Appender ingest isn't stalled by frequent checkpoints. `preserve_insertion_order=false` is always set — telemetry reads all have explicit ORDER BY, and dropping the guarantee lets DuckDB parallelize bulk loads and large scans with less memory. The read pool is bounded (`SetMaxOpenConns(duckDBMaxReadConns)`) since each DuckDB connection can use all threads + its own query memory; Appender writes use their own `DuckDBConnector.Connect()` connections and bypass that cap. Exposes `db.DuckDBConnector` (needed for the Appender).
-- **Writes use the Appender API**, not `INSERT` (`duckdb.NewAppenderFromConn(conn, "", table)` → `AppendRow(...)` → `Close()` flushes). Upserts still go through `ExecContext` with `ON CONFLICT`. The Appender rejects typed `*string` for nullable VARCHAR — use `nullableString()` in `backend/app/repositories/telemetry/duckdb/helpers.go` (returns untyped `nil` or the dereferenced value).
+- **Opened in** `backend/app/db/db_telemetry_duckdb.go`: telemetry path is the SQLite path with `.db` swapped for `_telemetry.duckdb`. By default DuckDB auto-tunes to the host; `DUCKDB_MEMORY_LIMIT`/`DUCKDB_THREADS`/`DUCKDB_CHECKPOINT_THRESHOLD` (passed through as DSN config options) let operators cap memory/threads so a memory-capped container doesn't read the host's RAM and OOM-kill the backend, and tune the WAL checkpoint threshold (backend default 256MB; DuckDB's own default of 16MB stalls sustained Appender ingest with frequent checkpoints). `preserve_insertion_order=false` is always set — telemetry reads all have explicit ORDER BY, and dropping the guarantee lets DuckDB parallelize bulk loads and large scans with less memory. The read pool is bounded (`SetMaxOpenConns(duckDBMaxReadConns)`) since each DuckDB connection can use all threads + its own query memory; Appender writes use their own `DuckDBConnector.Connect()` connections and bypass that cap. Exposes `db.DuckDBConnector` (needed for the Appender).
+- **Writes use the Appender API**, not `INSERT` (`duckdb.NewAppenderFromConn(conn, "", table)` → `AppendRow(...)` → `Close()` flushes). The five hot tables (`spans`, `metric_points`, `log_records`, `endpoints`, `tasks`) additionally go through a **background write batcher** (`telemetry/duckdb/writer.go`, started via `telemetry.StartWriters` in `cmd/run.go`): the request goroutine converts models to rows and enqueues onto a bounded per-table queue, one writer goroutine per table owns a persistent connection + Appender and flushes at `DUCKDB_WRITE_FLUSH_ROWS`/`DUCKDB_WRITE_FLUSH_INTERVAL_MS`. A 200 therefore means "accepted into the queue" (OTel-Collector-style ack-before-durable, loss window ≤ one flush); a full queue returns `db.ErrIngestQueueFull`, which controllers map to 503 + Retry-After (`middleware.AbortIngestInsertError`; the Retry-After hint tracks the configured queue wait) — overload is shed before the ack, acked rows are never silently dropped. `telemetry.FlushWriters` is a barrier used by the notification event evaluator (it reads back just-ingested rows) and tests; when writers aren't started (tests), `InsertAsync` falls back to the synchronous per-request appender. The other tables stay synchronous: `exception_stack_traces` (evaluator read-back), `sessions`/`profiling_stacks` (ordering-sensitive upserts), and the low-volume rest. Upserts still go through `ExecContext` with `ON CONFLICT`. The Appender rejects typed `*string` for nullable VARCHAR — use `nullableString()` in `backend/app/repositories/telemetry/duckdb/helpers.go` (returns untyped `nil` or the dereferenced value).
 - **Write-path observability:** a row the Appender rejects is dropped rather than failing the whole frame (the SQLite backend 500s instead), so a poison row cannot wedge the SDK's retry loop. Every drop increments a per-table counter (`db.RecordTelemetryRowDropped`) and fires a rate-limited (1/min per table) `traceway.CaptureException`; Appender flush/connect failures still propagate to the request (500, SDK retries) and increment an insert-failure counter. `GET /api/health/deep` (App auth, all telemetry backends) exposes `telemetryBackend`, `droppedRows` per table, `droppedRowsTotal`, `insertFailures`, and on DuckDB an `engine` object (db/WAL file bytes, `duckdb_memory()` usage, read-pool in-use/wait stats) alongside its existing ClickHouse fields; it 503s only when the configured telemetry backend is ClickHouse and CH is unreachable (the embedded backends answer 200 with `chReachable:false`). The benchmark loadgen polls it before/after every ramp step and fails any step whose drop delta is nonzero; read-probe fills record cumulative `droppedRows` per fill level. When `MONITORING_TRACEWAY_URL` is set, `monitoring.StartTelemetryDBReporter` also emits `traceway.duckdb.*` metrics every 10s: `rows_dropped.delta`, `insert_failures.delta`, `db_size_mb`, `wal_size_mb`, `memory_used_mb`, `read_pool.in_use`, `read_pool.wait_count.delta`, `read_pool.wait_ms.delta`. The hourly retention worker issues a `CHECKPOINT` after its deletes so retention actually reclaims disk (DuckDB otherwise defers reclamation to the WAL checkpoint threshold).
 - **`lit` placeholders:** `db.Driver` stays `lit.SQLite`, which emits `?` — DuckDB accepts these, so no separate driver was needed for reads.
 - **Migrations:** `backend/app/migrations/duckdb_telemetry/` (mirrors `sqlite_telemetry/` table-for-table; integer columns are `BIGINT`, JSON is `VARCHAR`, no secondary indexes since it's columnar).

backend/app/config/config.go

@@ -14,9 +14,14 @@ type Cfg struct {
 	PostgresSSLMode  string
 	SQLitePath       string
 
-	DuckDBMemoryLimit         string
-	DuckDBThreads             string
-	DuckDBCheckpointThreshold string
+	DuckDBMemoryLimit          string
+	DuckDBThreads              string
+	DuckDBCheckpointThreshold  string
+	DuckDBWriteQueueRows       string
+	DuckDBWriteFlushRows       string
+	DuckDBWriteFlushIntervalMS string
+	DuckDBWriteWriters         string
+	DuckDBWriteQueueWaitMS     string
 
 	ClickhouseServer   string
 	ClickhouseDatabase string
@@ -61,6 +66,7 @@ type Cfg struct {
 	MonitoringTracewayURL string
 	APIOnly               string
 	Ports                 string
+	PprofPort             string
 	TurnstileSecretKey    string
 
 	GoogleClientID     string
@@ -102,9 +108,14 @@ func LoadFromEnv() *Cfg {
 		PostgresSSLMode:  os.Getenv("POSTGRES_SSLMODE"),
 		SQLitePath:       os.Getenv("SQLITE_PATH"),
 
-		DuckDBMemoryLimit:         os.Getenv("DUCKDB_MEMORY_LIMIT"),
-		DuckDBThreads:             os.Getenv("DUCKDB_THREADS"),
-		DuckDBCheckpointThreshold: os.Getenv("DUCKDB_CHECKPOINT_THRESHOLD"),
+		DuckDBMemoryLimit:          os.Getenv("DUCKDB_MEMORY_LIMIT"),
+		DuckDBThreads:              os.Getenv("DUCKDB_THREADS"),
+		DuckDBCheckpointThreshold:  os.Getenv("DUCKDB_CHECKPOINT_THRESHOLD"),
+		DuckDBWriteQueueRows:       os.Getenv("DUCKDB_WRITE_QUEUE_ROWS"),
+		DuckDBWriteFlushRows:       os.Getenv("DUCKDB_WRITE_FLUSH_ROWS"),
+		DuckDBWriteFlushIntervalMS: os.Getenv("DUCKDB_WRITE_FLUSH_INTERVAL_MS"),
+		DuckDBWriteWriters:         os.Getenv("DUCKDB_WRITE_WRITERS"),
+		DuckDBWriteQueueWaitMS:     os.Getenv("DUCKDB_WRITE_QUEUE_WAIT_MS"),
 
 		ClickhouseServer:   os.Getenv("CLICKHOUSE_SERVER"),
 		ClickhouseDatabase: os.Getenv("CLICKHOUSE_DATABASE"),
@@ -149,6 +160,7 @@ func LoadFromEnv() *Cfg {
 		MonitoringTracewayURL: os.Getenv("MONITORING_TRACEWAY_URL"),
 		APIOnly:               os.Getenv("API_ONLY"),
 		Ports:                 os.Getenv("PORTS"),
+		PprofPort:             os.Getenv("PPROF_PORT"),
 		TurnstileSecretKey:    os.Getenv("TURNSTILE_SECRET_KEY"),
 
 		GoogleClientID:     os.Getenv("GOOGLE_CLIENT_ID"),

backend/app/controllers/clientcontrollers/client.controller.go

@@ -305,7 +305,7 @@ func (e clientController) Report(c *gin.Context) {
 		err := telemetry.EndpointRepository.InsertAsync(c, endpointsToInsert)
 		insertSpan.End()
 		if err != nil {
-			c.AbortWithError(500, traceway.NewStackTraceErrorf("error inserting endpointsToInsert: %w", err))
+			middleware.AbortIngestInsertError(c, err, "endpointsToInsert")
 			return
 		}
 	}
@@ -315,7 +315,7 @@ func (e clientController) Report(c *gin.Context) {
 		err := telemetry.TaskRepository.InsertAsync(c, tasksToInsert)
 		insertSpan.End()
 		if err != nil {
-			c.AbortWithError(500, traceway.NewStackTraceErrorf("error inserting tasksToInsert: %w", err))
+			middleware.AbortIngestInsertError(c, err, "tasksToInsert")
 			return
 		}
 	}
@@ -344,7 +344,7 @@ func (e clientController) Report(c *gin.Context) {
 		err := telemetry.MetricPointRepository.InsertAsync(c, metricPointsToInsert)
 		insertSpan.End()
 		if err != nil {
-			c.AbortWithError(500, traceway.NewStackTraceErrorf("error inserting metricPointsToInsert: %w", err))
+			middleware.AbortIngestInsertError(c, err, "metricPointsToInsert")
 			return
 		}
 
@@ -357,7 +357,7 @@ func (e clientController) Report(c *gin.Context) {
 	spanInsertSpan.End()
 
 	if err != nil {
-		c.AbortWithError(500, traceway.NewStackTraceErrorf("error inserting spansToInsert: %w", err))
+		middleware.AbortIngestInsertError(c, err, "spansToInsert")
 		return
 	}
 

backend/app/controllers/health_deep.go

@@ -36,6 +36,7 @@ type HealthDeepResponse struct {
 	DroppedRowsTotal uint64                   `json:"droppedRowsTotal"`
 	InsertFailures   uint64                   `json:"insertFailures"`
 	IngestRejected   uint64                   `json:"ingestRejected"`
+	WriteQueueDepth  map[string]int           `json:"writeQueueDepth,omitempty"`
 	Engine           *db.TelemetryEngineStats `json:"engine,omitempty"`
 }
 
@@ -51,6 +52,9 @@ func (h healthDeepController) Get(c *gin.Context) {
 	resp.DroppedRowsTotal = droppedTotal
 	resp.InsertFailures = insertFailures
 	resp.IngestRejected = db.GetIngestRejects()
+	if depths, ok := db.GetTelemetryWriteQueueDepths(); ok {
+		resp.WriteQueueDepth = depths
+	}
 	if engine, ok := db.GetTelemetryEngineStats(c.Request.Context()); ok {
 		resp.Engine = &engine
 	}

backend/app/controllers/otelcontrollers/codec.go

@@ -1,11 +1,13 @@
 package otelcontrollers
 
 import (
+	"bytes"
 	"compress/gzip"
 	"fmt"
 	"io"
 	"net/http"
 	"strings"
+	"sync"
 
 	"github.com/gin-gonic/gin"
 	collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
@@ -18,17 +20,48 @@ import (
 
 const maxBodySize = 10 * 1024 * 1024 // 10MB
 
-func readBody(c *gin.Context) ([]byte, error) {
+// The decode path runs for every OTLP request; pooling the gzip reader and
+// the decompressed-body buffer removes a reader setup and the io.ReadAll
+// realloc ladder per request. Pooled memory never escapes this file: the
+// body bytes are only valid until release() and proto/protojson unmarshal
+// copies what it keeps.
+var (
+	gzipReaderPool sync.Pool
+	bodyBufPool    = sync.Pool{New: func() any { return new(bytes.Buffer) }}
+)
+
+func readBody(c *gin.Context) (body []byte, release func(), err error) {
 	var reader io.Reader = c.Request.Body
 	if strings.EqualFold(c.GetHeader("Content-Encoding"), "gzip") {
-		gr, err := gzip.NewReader(c.Request.Body)
+		var gr *gzip.Reader
+		if pooled, ok := gzipReaderPool.Get().(*gzip.Reader); ok {
+			gr = pooled
+			err = gr.Reset(c.Request.Body)
+		} else {
+			gr, err = gzip.NewReader(c.Request.Body)
+		}
 		if err != nil {
-			return nil, fmt.Errorf("failed to create gzip reader: %w", err)
+			if gr != nil {
+				gzipReaderPool.Put(gr)
+			}
+			return nil, nil, fmt.Errorf("failed to create gzip reader: %w", err)
 		}
-		defer gr.Close()
+		// The body is fully consumed below, so the reader can go back to
+		// the pool before returning.
+		defer func() {
+			gr.Close()
+			gzipReaderPool.Put(gr)
+		}()
 		reader = gr
 	}
-	return io.ReadAll(io.LimitReader(reader, maxBodySize))
+
+	buf := bodyBufPool.Get().(*bytes.Buffer)
+	buf.Reset()
+	if _, err := buf.ReadFrom(io.LimitReader(reader, maxBodySize)); err != nil {
+		bodyBufPool.Put(buf)
+		return nil, nil, err
+	}
+	return buf.Bytes(), func() { bodyBufPool.Put(buf) }, nil
 }
 
 func isProtobuf(c *gin.Context) bool {
@@ -37,10 +70,11 @@ func isProtobuf(c *gin.Context) bool {
 }
 
 func decodeTraceRequest(c *gin.Context) (*coltracepb.ExportTraceServiceRequest, int, error) {
-	body, err := readBody(c)
+	body, release, err := readBody(c)
 	if err != nil {
 		return nil, 0, err
 	}
+	defer release()
 	req := &coltracepb.ExportTraceServiceRequest{}
 	if isProtobuf(c) {
 		if err := proto.Unmarshal(body, req); err != nil {
@@ -55,10 +89,11 @@ func decodeTraceRequest(c *gin.Context) (*coltracepb.ExportTraceServiceRequest,
 }
 
 func decodeMetricsRequest(c *gin.Context) (*colmetricspb.ExportMetricsServiceRequest, int, error) {
-	body, err := readBody(c)
+	body, release, err := readBody(c)
 	if err != nil {
 		return nil, 0, err
 	}
+	defer release()
 	req := &colmetricspb.ExportMetricsServiceRequest{}
 	if isProtobuf(c) {
 		if err := proto.Unmarshal(body, req); err != nil {
@@ -95,12 +130,15 @@ func writeMetricsResponse(c *gin.Context) {
 }
 
 func decodeProfilesPayload(c *gin.Context) ([]byte, int, error) {
-	body, err := readBody(c)
+	body, release, err := readBody(c)
 	if err != nil {
 		return nil, 0, err
 	}
+	defer release()
 	if isProtobuf(c) {
-		return body, len(body), nil
+		// The payload escapes to the profiling pipeline, so it must own its
+		// bytes — the pooled buffer is reused after release.
+		return bytes.Clone(body), len(body), nil
 	}
 	req := &colprofilespb.ExportProfilesServiceRequest{}
 	if err := protojson.Unmarshal(body, req); err != nil {
@@ -125,10 +163,11 @@ func writeProfilesResponse(c *gin.Context) {
 }
 
 func decodeLogsRequest(c *gin.Context) (*collogspb.ExportLogsServiceRequest, int, error) {
-	body, err := readBody(c)
+	body, release, err := readBody(c)
 	if err != nil {
 		return nil, 0, err
 	}
+	defer release()
 	req := &collogspb.ExportLogsServiceRequest{}
 	if isProtobuf(c) {
 		if err := proto.Unmarshal(body, req); err != nil {