k8s workers: bound pre-session memory (explicit memory_limit, GOMEMLIMIT, threads at spawn)

controlplane/control.go

@@ -1403,17 +1403,7 @@ func (cp *ControlPlane) workerDuckDBLimits(profile *WorkerProfile) (memLimit str
 	}
 
 	if memReq != "" {
-		memBytes := parseK8sMemory(memReq)
-		if memBytes > 0 {
-			duckdbBytes := memBytes * 3 / 4 // 75% of worker memory for DuckDB
-			const gb = 1024 * 1024 * 1024
-			const mb = 1024 * 1024
-			if duckdbBytes >= gb {
-				memLimit = fmt.Sprintf("%dGB", duckdbBytes/gb)
-			} else {
-				memLimit = fmt.Sprintf("%dMB", duckdbBytes/mb)
-			}
-		}
+		memLimit = duckdbMemoryLimitForPodMemory(parseK8sMemory(memReq))
 	}
 
 	if cpuReq != "" {
@@ -1423,6 +1413,26 @@ func (cp *ControlPlane) workerDuckDBLimits(profile *WorkerProfile) (memLimit str
 	return memLimit, threads
 }
 
+// duckdbMemoryLimitForPodMemory formats the DuckDB memory_limit for a worker
+// pod of the given memory size: 75% of the pod, leaving the remainder as
+// headroom for everything memory_limit does NOT govern (DuckDB C++ catalog
+// objects, postgres_scanner/libpq result buffers, the Go runtime, page cache).
+// Returns "" when memBytes is 0/unparseable (DuckDB then auto-detects).
+// Shared between session sizing (workerDuckDBLimits) and the spawn-time
+// DUCKGRES_MEMORY_LIMIT env so the two can never disagree.
+func duckdbMemoryLimitForPodMemory(memBytes uint64) string {
+	if memBytes == 0 {
+		return ""
+	}
+	duckdbBytes := memBytes * 3 / 4 // 75% of worker memory for DuckDB
+	const gb = 1024 * 1024 * 1024
+	const mb = 1024 * 1024
+	if duckdbBytes >= gb {
+		return fmt.Sprintf("%dGB", duckdbBytes/gb)
+	}
+	return fmt.Sprintf("%dMB", duckdbBytes/mb)
+}
+
 // workerBillingSize returns the provisioned worker pod size for compute-usage
 // billing, in milli-units (millicores, MiB). It mirrors workerDuckDBLimits's
 // source-of-truth selection: a non-default profile sizes from the profile's pod

controlplane/k8s_pool_spawn.go

@@ -74,6 +74,10 @@ func (p *K8sWorkerPool) spawnWorker(ctx context.Context, id int, image string, p
 		podLabels["duckgres/org"] = p.orgID
 	}
 
+	// Resolve the pod's resource shape once: the container Resources AND the
+	// memory-hygiene env vars below must derive from the same values.
+	workerResources := p.workerResourcesForProfile(profile)
+
 	// Build pod spec
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -152,7 +156,7 @@ func (p *K8sWorkerPool) spawnWorker(ctx context.Context, id int, image string, p
 					SecurityContext: &corev1.SecurityContext{
 						AllowPrivilegeEscalation: boolPtr(false),
 					},
-					Resources: p.workerResourcesForProfile(profile),
+					Resources: workerResources,
 				},
 			},
 		},
@@ -162,6 +166,18 @@ func (p *K8sWorkerPool) spawnWorker(ctx context.Context, id int, image string, p
 		Value: "true",
 	})
 
+	// Pre-session memory hygiene. Without an explicit DUCKGRES_MEMORY_LIMIT the
+	// worker's ConfigureMainDB falls back to sysinfo.AutoMemoryLimit(), which
+	// reads the NODE's /proc/meminfo — so all pre-session work (DuckLake
+	// ATTACH, activation, warmup, controlDB) runs with a memory_limit sized to
+	// the node, not the pod cgroup. Pass the pod-derived limit (same 75% the
+	// per-session SET uses, via duckdbMemoryLimitForPodMemory) and the pod's
+	// CPU count so the base DB is correctly bounded from process start.
+	// GOMEMLIMIT (read by the Go runtime directly) gives the Go side a soft
+	// ceiling at 1/8 of the pod so GC pushes back before the cgroup OOM-kills;
+	// DuckDB's buffer manager is unaffected (C allocations are untracked by Go).
+	pod.Spec.Containers[0].Env = append(pod.Spec.Containers[0].Env, workerMemoryHygieneEnv(workerResources)...)
+
 	// Stamp every log line with pod and node identifiers via the Downward API.
 	pod.Spec.Containers[0].Env = append(pod.Spec.Containers[0].Env,
 		corev1.EnvVar{
@@ -621,6 +637,29 @@ func (p *K8sWorkerPool) workerResources() corev1.ResourceRequirements {
 	return p.workerResourcesForProfile(WorkerProfile{})
 }
 
+// workerMemoryHygieneEnv derives the memory/thread env vars for a worker pod
+// from its resolved resource requests. See the call site comment in
+// spawnWorker for why these must be set before any session exists.
+func workerMemoryHygieneEnv(res corev1.ResourceRequirements) []corev1.EnvVar {
+	var env []corev1.EnvVar
+	if mem, ok := res.Requests[corev1.ResourceMemory]; ok {
+		memBytes := uint64(mem.Value())
+		if lim := duckdbMemoryLimitForPodMemory(memBytes); lim != "" {
+			env = append(env, corev1.EnvVar{Name: "DUCKGRES_MEMORY_LIMIT", Value: lim})
+		}
+		if goLimitMiB := memBytes / 8 / (1 << 20); goLimitMiB > 0 {
+			env = append(env, corev1.EnvVar{Name: "GOMEMLIMIT", Value: fmt.Sprintf("%dMiB", goLimitMiB)})
+		}
+	}
+	if cpu, ok := res.Requests[corev1.ResourceCPU]; ok {
+		// Value() rounds fractional cores up, so a "500m" worker gets 1 thread.
+		if threads := cpu.Value(); threads > 0 {
+			env = append(env, corev1.EnvVar{Name: "DUCKGRES_THREADS", Value: strconv.FormatInt(threads, 10)})
+		}
+	}
+	return env
+}
+
 // workerResourcesForProfile builds the pod resource requirements for a worker of
 // the given profile: the profile's CPU/Memory when set, otherwise the pool-global
 // request (today's behavior). Requests are mirrored to Limits so the pod is

controlplane/k8s_pool_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/posthog/duckgres/server"
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -3001,7 +3002,9 @@ func assertSpawnedWorkerPod(t *testing.T, pod *corev1.Pod) {
 	foundTLSCertEnv := false
 	foundTLSKeyEnv := false
 	foundMaxSessionsEnv := false
+	envByName := map[string]string{}
 	for _, env := range c.Env {
+		envByName[env.Name] = env.Value
 		if env.Name == "DUCKGRES_DUCKDB_TOKEN" && env.ValueFrom != nil &&
 			env.ValueFrom.SecretKeyRef != nil &&
 			env.ValueFrom.SecretKeyRef.Name == "test-secret-duckgres-worker-test-cp-0" {
@@ -3033,6 +3036,22 @@ func assertSpawnedWorkerPod(t *testing.T, pod *corev1.Pod) {
 		t.Fatal("expected DUCKGRES_DUCKDB_MAX_SESSIONS=1 (one query session per worker)")
 	}
 
+	// Pre-session memory hygiene: the worker process must know the POD's
+	// memory shape from the start. Without these, ConfigureMainDB sizes the
+	// base DB off the NODE's /proc/meminfo (sysinfo.AutoMemoryLimit) and all
+	// pre-session work (DuckLake ATTACH, activation) runs effectively
+	// unbounded, and the Go runtime has no ceiling at all.
+	// Default pool shape is 16Gi/8cpu -> 75% = 12GB, GOMEMLIMIT = 1/8 pod.
+	if got := envByName["DUCKGRES_MEMORY_LIMIT"]; got != "12GB" {
+		t.Fatalf("expected DUCKGRES_MEMORY_LIMIT=12GB (75%% of 16Gi pod), got %q", got)
+	}
+	if got := envByName["DUCKGRES_THREADS"]; got != "8" {
+		t.Fatalf("expected DUCKGRES_THREADS=8 (pod CPU request), got %q", got)
+	}
+	if got := envByName["GOMEMLIMIT"]; got != "2048MiB" {
+		t.Fatalf("expected GOMEMLIMIT=2048MiB (1/8 of 16Gi pod), got %q", got)
+	}
+
 	if len(pod.Spec.Volumes) == 0 {
 		t.Fatal("expected configmap volume")
 	}
@@ -4614,3 +4633,45 @@ func TestWorkerDoNotDisruptLifecycle(t *testing.T) {
 		t.Fatalf("do-not-disrupt annotation = %q,%v after protect, want \"true\"", v, ok)
 	}
 }
+
+func TestWorkerMemoryHygieneEnv(t *testing.T) {
+	mk := func(cpu, mem string) corev1.ResourceRequirements {
+		req := corev1.ResourceList{}
+		if cpu != "" {
+			req[corev1.ResourceCPU] = resource.MustParse(cpu)
+		}
+		if mem != "" {
+			req[corev1.ResourceMemory] = resource.MustParse(mem)
+		}
+		return corev1.ResourceRequirements{Requests: req}
+	}
+	cases := []struct {
+		name     string
+		res      corev1.ResourceRequirements
+		expected map[string]string
+	}{
+		{"prod-like 15/120Gi", mk("15", "120Gi"), map[string]string{
+			"DUCKGRES_MEMORY_LIMIT": "90GB", "GOMEMLIMIT": "15360MiB", "DUCKGRES_THREADS": "15"}},
+		{"dev 15/64Gi", mk("15", "64Gi"), map[string]string{
+			"DUCKGRES_MEMORY_LIMIT": "48GB", "GOMEMLIMIT": "8192MiB", "DUCKGRES_THREADS": "15"}},
+		{"sub-GB pod formats MB", mk("500m", "512Mi"), map[string]string{
+			"DUCKGRES_MEMORY_LIMIT": "384MB", "GOMEMLIMIT": "64MiB", "DUCKGRES_THREADS": "1"}},
+		{"no requests -> no env", corev1.ResourceRequirements{}, map[string]string{}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := map[string]string{}
+			for _, e := range workerMemoryHygieneEnv(tc.res) {
+				got[e.Name] = e.Value
+			}
+			if len(got) != len(tc.expected) {
+				t.Fatalf("env mismatch: got %v want %v", got, tc.expected)
+			}
+			for k, v := range tc.expected {
+				if got[k] != v {
+					t.Fatalf("%s = %q, want %q (all: %v)", k, got[k], v, got)
+				}
+			}
+		})
+	}
+}

tests/e2e-mw-dev/harness.sh

@@ -852,6 +852,19 @@ assert_worker_pod() { # org
   dmem="$(k get pod "$pod" -o jsonpath="${WORKER_C}.resources.requests.memory}")"
   [ "$dcpu" = "750m" ] || fail "default worker $pod requests.cpu='$dcpu' want '750m' (pool default; BestEffort regression?)"
   [ "$dmem" = "1536Mi" ] || fail "default worker $pod requests.memory='$dmem' want '1536Mi'"
+
+  # Pre-session memory hygiene env (workerMemoryHygieneEnv): the CP must stamp
+  # a pod-derived DuckDB memory_limit, a Go soft memory ceiling, and the thread
+  # count at spawn — otherwise the worker sizes its base DB off the NODE's
+  # /proc/meminfo and all pre-session work (DuckLake ATTACH, activation) runs
+  # effectively unbounded. Values derive from the 750m/1536Mi pool default:
+  # 75% of 1536Mi floors to 1GB; GOMEMLIMIT is 1/8 pod = 192MiB; 750m -> 1.
+  dml="$(k get pod "$pod" -o jsonpath="${WORKER_C}.env[?(@.name==\"DUCKGRES_MEMORY_LIMIT\")].value}")"
+  [ "$dml" = "1GB" ] || fail "default worker $pod DUCKGRES_MEMORY_LIMIT='$dml' want '1GB' (75% of 1536Mi, GB-floored)"
+  gml="$(k get pod "$pod" -o jsonpath="${WORKER_C}.env[?(@.name==\"GOMEMLIMIT\")].value}")"
+  [ "$gml" = "192MiB" ] || fail "default worker $pod GOMEMLIMIT='$gml' want '192MiB' (1/8 of 1536Mi)"
+  thr="$(k get pod "$pod" -o jsonpath="${WORKER_C}.env[?(@.name==\"DUCKGRES_THREADS\")].value}")"
+  [ "$thr" = "1" ] || fail "default worker $pod DUCKGRES_THREADS='$thr' want '1' (750m rounds up)"
 }
 
 # ---- worker sizing (TTL-pool model) ---------------------------------------