feat(server): add --statement-timeout to bound wedged statements

CLAUDE.md

@@ -130,6 +130,11 @@ Key CLI flags for control-plane mode:
 - `--process-min-workers N` / `--process-max-workers N`
 - `--process-retire-on-session-end`
 - `--worker-queue-timeout DURATION` / `--worker-idle-timeout DURATION`
+- `--statement-timeout DURATION` — cancels any single statement that runs longer than this; `0`/empty (the default) leaves statements **unbounded**, which is today's behavior. Blast-radius containment only: nothing else in the stack bounds a wedged statement (DuckDB has no statement timeout, client `statement_timeout` is an ignored SET, the cnpg shards run `idle_in_transaction_session_timeout=0`, pgwire clients wait forever), so one engine deadlock cost a tenant 5h33m of their hourly refresh pipeline on 2026-09-16. The deadline rides on the STATEMENT context, never the connection context — a timeout ends one statement, not the session — and surfaces as `57014` with PostgreSQL's exact wording `canceling statement due to statement timeout` (drivers string-match it).
+  - **Every execution path must acquire the statement context through `queryContext`/`queryContextInner`.** That is the ONLY place `RegisterQuery` is called, so a path that skips it is reachable by neither the timeout nor a pgwire `CancelRequest`. Extended-protocol `Execute` used to skip it and called the context-LESS `executor.Query`/`Exec` (which run on `context.Background()` — see `PinnedExecutor`), leaving prepared statements — what pgx, psycopg3 and JDBC send — unbounded and uncancellable. Regression: `TestStatementTimeoutReachesExtendedProtocolExecute`. If you add an execution path, route it through `queryContext` and the `*Context` executor methods.
+  - **Timeout classification is driven by the ERROR and gated on the timeout being configured** (`statementTimedOut(err)`), never by stored per-connection state. An earlier design recorded the live statement context on the connection; it went sticky — once any statement timed out, later unrelated failures were reported as timeouts and their logging skipped. The gate also keeps a deployment with the feature OFF byte-for-byte identical, so internal deadlines (attach, exec, worker gRPC) do not start surfacing as `57014`. With the feature ON an internal deadline racing a statement can still be labelled a statement timeout — an accepted narrowing.
+  - **A cursor is bounded over its whole lifetime, not per `FETCH`**: `queryContextForCursor` takes ONE context at `DECLARE` and holds it across every `FETCH`, so a cursor open longer than the timeout dies — stricter than PostgreSQL. Deliberate (an unbounded cursor also pins a worker), but size the timeout with long-lived cursors in mind.
+  - Caveat: an engine that ignores cancellation (duckdb/duckdb#24961) will not stop work; the client is freed with a clear error but the worker can stay wedged — retiring a timed-out worker is a follow-up that must respect the destroy-before-reuse ordering in `SessionManager.DestroySession`.
 - `--idle-timeout DURATION` — connection idle timeout: a client connection with no traffic for this long is closed and its worker released to hot-idle (in control-plane mode an idle connection otherwise pins a worker forever). The close is preceded by a `FATAL` 57P05 ErrorResponse naming the effective timeout (matching PostgreSQL's `idle_session_timeout`), so clients see a reap, not a network fault. **Control-plane default is `5m`** (`server.DefaultControlPlaneIdleTimeout`; standalone defaults to `24h`); a negative value disables it. `server.New` applies the standalone default, so the control plane sets it explicitly before `InitMinimalServer` (which skips that defaulting).
 - `--memory-budget SIZE` (default 75% RAM) / `--memory-rebalance`
 - `--socket-dir /path` (process backend)

README.md

@@ -674,6 +674,8 @@ Options:
   -threads int             DuckDB threads per session
   -process-isolation       Enable process isolation (spawn child process per connection)
   -idle-timeout string     Connection idle timeout (e.g., '30m', '1h', '-1' to disable)
+  -statement-timeout string
+                          Cancel any single statement running longer than this (e.g., '60m'); empty/'0' = unbounded
   -mode string             Run mode: standalone (default), control-plane, duckdb-service, or reshard-runner
   -process-min-workers int Pre-warm process worker count at startup (control-plane mode, default 0)
   -process-max-workers int Max process workers, 0=auto-derived (control-plane mode)

configloader/file_config.go

@@ -25,6 +25,7 @@ type FileConfig struct {
 	ProcessIsolation     bool                `yaml:"process_isolation"`
 	IdleTimeout          string              `yaml:"idle_timeout"`
 	SessionInitTimeout   string              `yaml:"session_init_timeout"`
+	StatementTimeout     string              `yaml:"statement_timeout"`
 	MemoryLimit          string              `yaml:"memory_limit"`
 	Threads              int                 `yaml:"threads"`
 	MemoryBudget         string              `yaml:"memory_budget"`

configresolve/cliflags.go

@@ -31,6 +31,7 @@ func RegisterCLIInputsFlags(fs *flag.FlagSet) func() CLIInputs {
 	processIsolation := fs.Bool("process-isolation", false, "Enable process isolation (spawn child process per connection)")
 	idleTimeout := fs.String("idle-timeout", "", "Connection idle timeout: close a connection idle (no traffic) this long, freeing its worker (e.g., '30m', '1h', '-1s' to disable). Default 24h standalone, 5m control-plane where idle connections pin a worker (env: DUCKGRES_IDLE_TIMEOUT)")
 	sessionInitTimeout := fs.String("session-init-timeout", "", "Session startup metadata/probe timeout (e.g., '10s', '30s') (env: DUCKGRES_SESSION_INIT_TIMEOUT)")
+	statementTimeout := fs.String("statement-timeout", "", "Cancel any single statement that runs longer than this (e.g., '60m'); empty or '0' leaves statements unbounded (env: DUCKGRES_STATEMENT_TIMEOUT)")
 	memoryLimit := fs.String("memory-limit", "", "DuckDB memory_limit per session (e.g., '4GB') (env: DUCKGRES_MEMORY_LIMIT)")
 	threads := fs.Int("threads", 0, "DuckDB threads per session (env: DUCKGRES_THREADS)")
 	memoryBudget := fs.String("memory-budget", "", "Total memory for all DuckDB sessions (e.g., '24GB') (env: DUCKGRES_MEMORY_BUDGET)")
@@ -85,6 +86,7 @@ func RegisterCLIInputsFlags(fs *flag.FlagSet) func() CLIInputs {
 		cli.ProcessIsolation = *processIsolation
 		cli.IdleTimeout = *idleTimeout
 		cli.SessionInitTimeout = *sessionInitTimeout
+		cli.StatementTimeout = *statementTimeout
 		cli.MemoryLimit = *memoryLimit
 		cli.Threads = *threads
 		cli.MemoryBudget = *memoryBudget

configresolve/cliflags_test.go

@@ -64,6 +64,7 @@ func fieldNameToFlagName(name string) string {
 		{"FilePersistence", "file-persistence"},
 		{"IdleTimeout", "idle-timeout"},
 		{"SessionInitTimeout", "session-init-timeout"},
+		{"StatementTimeout", "statement-timeout"},
 		{"MemoryLimit", "memory-limit"},
 		{"MemoryBudget", "memory-budget"},
 		{"MemoryRebalance", "memory-rebalance"},

configresolve/resolve.go

@@ -36,6 +36,7 @@ type CLIInputs struct {
 	ProcessIsolation            bool
 	IdleTimeout                 string
 	SessionInitTimeout          string
+	StatementTimeout            string
 	MemoryLimit                 string
 	Threads                     int
 	MemoryBudget                string
@@ -89,6 +90,7 @@ type Resolved struct {
 	ProcessMaxWorkers               int
 	ProcessRetireOnSessionEnd       bool
 	SessionInitTimeout              time.Duration
+	StatementTimeout                time.Duration
 	WorkerQueueTimeout              time.Duration
 	WorkerIdleTimeout               time.Duration
 	HandoverDrainTimeout            time.Duration
@@ -340,6 +342,13 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 				warn("Invalid idle_timeout duration: " + err.Error())
 			}
 		}
+		if fileCfg.StatementTimeout != "" {
+			if d, err := time.ParseDuration(fileCfg.StatementTimeout); err == nil {
+				cfg.StatementTimeout = d
+			} else {
+				warn("Invalid statement_timeout duration: " + err.Error())
+			}
+		}
 		if fileCfg.SessionInitTimeout != "" {
 			if d, err := time.ParseDuration(fileCfg.SessionInitTimeout); err == nil {
 				cfg.SessionInitTimeout = d
@@ -595,6 +604,13 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 			warn("Invalid DUCKGRES_CLIENT_IDLE_TIMEOUT_MAX: must be a positive duration")
 		}
 	}
+	if v := getenv("DUCKGRES_STATEMENT_TIMEOUT"); v != "" {
+		if d, err := time.ParseDuration(v); err == nil {
+			cfg.StatementTimeout = d
+		} else {
+			warn("Invalid DUCKGRES_STATEMENT_TIMEOUT duration: " + err.Error())
+		}
+	}
 	if v := getenv("DUCKGRES_SESSION_INIT_TIMEOUT"); v != "" {
 		if d, err := time.ParseDuration(v); err == nil {
 			cfg.SessionInitTimeout = d
@@ -942,6 +958,13 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 			warn("Invalid --idle-timeout duration: " + err.Error())
 		}
 	}
+	if cli.Set["statement-timeout"] {
+		if d, err := time.ParseDuration(cli.StatementTimeout); err == nil {
+			cfg.StatementTimeout = d
+		} else {
+			warn("Invalid --statement-timeout duration: " + err.Error())
+		}
+	}
 	if cli.Set["session-init-timeout"] {
 		if d, err := time.ParseDuration(cli.SessionInitTimeout); err == nil {
 			cfg.SessionInitTimeout = d

docs/plans/2026-09-16-warehouse-reliability-plan.md

@@ -0,0 +1,131 @@
+# Managed-warehouse reliability: root causes and fix plan (2026-09-16)
+
+Three distinct production failures were investigated on `mw-prod-us`. They were
+initially conflated; they are unrelated and have different fixes.
+
+---
+
+## 1. Portola sqlmesh `COMMIT ... SSL error: unexpected eof`  — FIX IN FLIGHT
+
+**Root cause (confirmed by source + measurement).**
+`information_schema.tables` is defined as `... FROM duckdb_tables()`, and
+`duckdb_tables()` calls `TableCatalogEntry::GetStorageInfo` for EVERY table in its
+row loop purely to fill `estimated_size`/`index_count`. In DuckLake that landed on
+`GetTableStats`, whose metadata query joins `ducklake_table_column_stats` and is
+filtered `WHERE table_id = <N>` — one metadata round-trip per table, shipping every
+column's min/max/extra_stats, to extract one integer (`record_count`). The stats
+cache is keyed `<next_file_id, table_id>`, so every commit that writes a file
+invalidates all of it.
+
+**Measured** (portola catalog: 4,497 tables / 516,944 column-stats rows):
+
+| | queries | rows | server-side |
+|---|---|---|---|
+| Current | 4,500 | 516,971 | 185 ms |
+| Batched, cardinality-only | 1 | 4,500 | 1 ms |
+
+Server execution is 185 ms — the cost is **not** database work, it is 4,500
+sequential round-trips. ~1.25 ms/query on a local socket (5.6s floor with no
+network); 4.4–9.1 ms/query through a pooler, which reproduces the observed
+20–41s listing exactly. Scales linearly with table count.
+
+The listing holds a read transaction open on portola's ATTACHed Postgres the
+whole time; their `portola_warehouse` role caps `idle_in_transaction_session_timeout`
+at 30s, so Postgres kills it and DuckDB reports SSL EOF on COMMIT.
+
+**Fix:** PostHog/ducklake#43 (base `posthog/v1.5.5`). Cardinality-only batch path;
+full `GetTableStats` untouched so query plans cannot change.
+
+**Remaining work:**
+- [ ] CI green on #43 (never built locally)
+- [ ] Time a real listing on a large catalog: confirm 20–41s → sub-second
+- [ ] Tag the fork; bump `DUCKLAKE_EXTENSION_TAG` in `Dockerfile` + `Dockerfile.worker`
+- [ ] e2e assertion per the repo testing contract
+- [ ] Promote to prod, then confirm EOFs stop
+
+**Fallback if that slips:** raise the idle-in-transaction window from our side.
+`PostHog/duckdb-postgres` already has `pg_idle_in_transaction_timeout_millis`, but
+it is (a) never set by duckgres and (b) only applied in `PostgresScanConnect` —
+NOT on the `PostgresTransaction` path that catalog listings actually use. The SET
+must be issued *inside* the transaction (PgBouncer transaction pooling discards
+session state between transactions, not within one), and the GUC is `USERSET`, so
+a client may raise it above a role default without touching the customer's DB.
+**It must raise only, never lower, and must preserve `0` (unlimited)** — the
+DuckLake metadata store runs unlimited today and a legitimate posthog data-import
+held an idle metadata transaction for 2h26m. A blanket 120s would have killed it.
+
+Portola's own PR (portolans/backend#9397) does the same thing via
+`ALTER ROLE portola_warehouse SET idle_in_transaction_session_timeout = '120s'`.
+That is a valid stopgap but needs their DBA, and does not survive catalog growth.
+
+---
+
+## 2. Portola 5.5h refresh wedge — in-engine DuckDB deadlock  — NOT FIXED
+
+**Root cause.** Worker `281869` ran
+`INSERT INTO sqlmesh__reports.reports__cohorts_dist_d7_daily...` — `cohorts_dist.sql`,
+"the expensive half: QUANTILE_CONT and percentile aggregations" over **18 GROUPING
+SETS**, a heavy blocking collapse. It did ~6 min of real work (09:44–09:50, 2 cores)
+then sat at **0.075 cores / ~380 B/s (TCP keepalive only) for ~5 hours**. All 62
+threads parked in futex/nanosleep, none in a socket read; gdb recovers only Go
+`runtime.futex` frames. Query duration logged **19,994,187 ms (5h33m)**, ended only
+by the GitHub Actions 6h job timeout.
+
+Worker was 20 min old (fresh credentials) and portola has zero `ExpiredToken` — this
+is not a credential problem. Matches upstream duckdb/duckdb#24961: blocking-collapse
+query hangs, `interrupt()` never honored, all threads park, CPU→0, needs SIGKILL.
+
+**Blast radius.** `sqlmesh-deploy.yml` uses concurrency group `sqlmesh-prod` with
+`cancel-in-progress: false`, so five consecutive hourly refreshes queued behind the
+wedge and were cancelled.
+
+**Plan:**
+- [ ] **Containment first** — a statement timeout in duckgres. Nothing bounds a wedged
+      statement today: no duckgres statement timeout, `idle_in_transaction_session_timeout=0`
+      on the shards, and sqlmesh/psycopg wait forever. This is the single highest-value
+      change for blast radius and is independent of the engine bug.
+- [ ] Track duckdb/duckdb#24961; ship an engine build with the fix
+- [ ] Workaround for portola: reduce grouping sets / split the model
+
+---
+
+## 3. PostHog Metabase failures — expired STS token on a pinned worker  — NOT FIXED
+
+**Root cause.** Worker `279924` started 18:16 and held ONE long-lived Metabase
+connection for ~19h. Per-tenant STS credentials expire (~1h), but worker-side
+refresh (`RefreshS3Secret` via `SessionPool.reuseExistingActivation`) only fires when
+a worker is **reused** for a new session or reclaimed from hot-idle — never
+periodically for a worker continuously busy with one session. The per-connection
+`StartCredentialRefresh` (5-min ticker) is deliberately not started on the
+remote/sharedDB path ("the pool manages it") — but the pool only refreshes on reuse.
+
+Result: 74 failures 00:04→13:35, **each ~256 s then fail**, with
+`HTTP GET ... ExpiredToken: The provided token has expired`. Self-resolved only when
+the worker was finally recycled.
+
+The 256 s is the httpfs retry budget duckgres sets to ride out S3 503 SlowDown
+(`applyHTTPFSRetryBudget`: retries=10, wait=500ms, backoff=2 ≈ 255s cumulative). That
+budget is being spent on a **non-retryable** auth error. The httpfs fork does have
+in-engine `RunWithCredentialRefresh` → `TryRefreshAuthParams`, but it only retries
+once and only if something committed fresh credentials — nothing did.
+
+**Plan:**
+- [ ] Refresh STS credentials for long-lived **active** sessions, not just on reuse
+- [ ] Do not spend the full retry budget on non-retryable auth 400s — fail fast or
+      force a secret refresh
+- [ ] Consider a max worker/session age so a pinned client connection cannot outlive
+      its credentials indefinitely
+
+---
+
+## Cross-cutting
+
+**Nothing bounds a hung or doomed statement anywhere in the stack.** Items 2 and 3
+are different bugs with the same amplifier. A statement timeout plus an
+idle-in-transaction bound would have turned a 5.5h outage into a bounded failure and
+a 256 s stall into an immediate error. Highest-leverage single change after #43.
+
+**Portola-side (their repo, no DBA needed):** 3,987 of 5,940 tables are dead sqlmesh
+dev environments. Invalidating unused envs so the hourly janitor drops them is what
+makes listings cheap permanently — the engine fix removes the per-table round-trip,
+but catalog size still drives everything else.

server/conn.go

@@ -134,6 +134,12 @@ type portalExec struct {
 	// finishProfiling runs after rows.Close has allowed a Flight DoGet trailer
 	// to arrive. A suspended portal retains it until its terminal Execute.
 	finishProfiling func()
+	// releaseStmtCtx cancels/unregisters the statement context whose deadline
+	// bounds this portal. A suspended portal's RowSet outlives the Execute that
+	// opened it, so — like finishProfiling — the portal owns the release until
+	// its terminal Execute. Cancelling at the opening Execute's return would
+	// tear down the rowset the next Execute resumes from.
+	releaseStmtCtx func()
 }
 
 // closeExec releases a suspended portal's open rowset (if any). Must be
@@ -147,6 +153,9 @@ func (p *portal) closeExec() {
 	if p.exec.finishProfiling != nil {
 		p.exec.finishProfiling()
 	}
+	if p.exec.releaseStmtCtx != nil {
+		p.exec.releaseStmtCtx()
+	}
 	p.exec = nil
 }
 
@@ -488,7 +497,16 @@ func (c *clientConn) queryContextInner(monitor bool) (context.Context, func()) {
 	// Carry the statement's ID to the engine. The worker stamps it on its own
 	// logs, so a client complaint can be followed from this connection into the
 	// pod that ran the statement.
-	ctx, cancel := context.WithCancel(wire.WithQueryID(c.ctx, c.currentQueryID()))
+	// The statement timeout rides on the statement context, not the connection
+	// context: a timeout must end one statement, not the session. That is also
+	// why statementTimedOut() cannot be answered from c.ctx -- see conn_errors.go.
+	var ctx context.Context
+	var cancel context.CancelFunc
+	if timeout := c.statementTimeout(); timeout > 0 {
+		ctx, cancel = context.WithTimeout(wire.WithQueryID(c.ctx, c.currentQueryID()), timeout)
+	} else {
+		ctx, cancel = context.WithCancel(wire.WithQueryID(c.ctx, c.currentQueryID()))
+	}
 	key := c.backendKey()
 	c.server.RegisterQuery(key, cancel)
 
@@ -1867,7 +1885,7 @@ func (c *clientConn) handleQuery(body []byte) (retErr error) {
 				errCode := classifyErrorCode(err)
 				errMsg := err.Error()
 				if c.isCallerCancellation(err) {
-					errMsg = "canceling statement due to user request"
+					errMsg = c.cancellationMessage(err)
 				} else {
 					c.logQueryError(query, err)
 				}

server/conn_cursor.go

@@ -306,7 +306,7 @@ func (c *clientConn) handleFetchCursor(query string, stmt *pg_query.FetchStmt) e
 			errMsg := err.Error()
 			if c.isCallerCancellation(err) {
 				errCode = "57014"
-				errMsg = "canceling statement due to user request"
+				errMsg = c.cancellationMessage(err)
 			}
 			c.sendError("ERROR", errCode, errMsg)
 			c.setTxError()
@@ -381,7 +381,7 @@ func (c *clientConn) handleFetchCursor(query string, stmt *pg_query.FetchStmt) e
 		errMsg := err.Error()
 		if c.isCallerCancellation(err) {
 			errCode = "57014"
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		}
 		c.sendError("ERROR", errCode, errMsg)
 		c.setTxError()
@@ -447,7 +447,7 @@ func (c *clientConn) handleFetchCursorExtended(p *portal) {
 	if cursor.rows == nil {
 		if err := c.openCursor(cursor); err != nil {
 			if c.isCallerCancellation(err) {
-				c.sendError("ERROR", "57014", "canceling statement due to user request")
+				c.sendError("ERROR", "57014", c.cancellationMessage(err))
 			} else {
 				c.sendError("ERROR", "42000", err.Error())
 			}
@@ -506,7 +506,7 @@ func (c *clientConn) handleFetchCursorExtended(p *portal) {
 
 	if err := cursor.rows.Err(); err != nil {
 		if c.isCallerCancellation(err) {
-			c.sendError("ERROR", "57014", "canceling statement due to user request")
+			c.sendError("ERROR", "57014", c.cancellationMessage(err))
 		} else {
 			c.sendError("ERROR", "42000", err.Error())
 		}