feat: add per-org memory admission limits

README.md

@@ -92,6 +92,8 @@ labels, aggregation rules, PromQL examples, and admission metric migration.
 | `duckgres_session_admission_queue_depth{org}` | Gauge | Local callers waiting after successful durable enqueue; sum across replicas |
 | `duckgres_session_admission_active_vcpus{org}` | Gauge | Requested vCPUs held by local live lease handles; cleanup-pending durable rows are excluded |
 | `duckgres_session_admission_limit_vcpus{org}` | Gauge | Config-reconciled effective org cap for active org stacks; zero means unlimited, max across replicas |
+| `duckgres_session_admission_active_memory_bytes{org}` | Gauge | Requested memory bytes held by local live lease handles; cleanup-pending durable rows are excluded |
+| `duckgres_session_admission_limit_memory_bytes{org}` | Gauge | Config-reconciled effective org memory cap for active org stacks; zero means unlimited, max across replicas |
 | `duckgres_session_admission_reclaim_pending` | Gauge | Activated cleanup intents awaiting or executing exact database reclamation |
 | `duckgres_session_admission_reclaim_attempts_total{outcome}` | Counter | Exact cleanup attempts by `success` or `error` outcome |
 | `duckgres_session_admission_reclaim_reservations_in_use` | Gauge | Cleanup-ownership slots held before enqueue, while queued or live, and during pending cleanup |
@@ -231,7 +233,7 @@ column cannot be added to a populated table).
 - [Performance Harness](docs/perf-harness-runbook.md): Local smoke and nightly operations for performance testing.
 - [Dev Scenario Runner](docs/runbooks/scenario-dev.md): Scheduled and manually dispatched scenario runs against the configured dev environment.
 - [Control Plane Rollout](docs/runbooks/control-plane-rollout.md): Zero-downtime deployment process for the control plane itself.
-- [Org Connection Admission](docs/runbooks/org-connection-admission.md): Global vCPU admission, exact cleanup ownership, failure recovery, and operational metrics.
+- [Org Connection Admission](docs/runbooks/org-connection-admission.md): Global CPU/memory admission, exact cleanup ownership, failure recovery, and operational metrics.
 - [Managed Warehouse Provisioning Recovery](docs/runbooks/managed-warehouse-provisioning-recovery.md): Diagnose a failed warehouse whose Duckling dependencies were repaired and verify automatic convergence back to ready.
 - [Managed Warehouse Deprovision](docs/runbooks/managed-warehouse-deprovision.md): Destructive teardown process for managed warehouse infrastructure and org cleanup.
 - [Resharding Operations](docs/runbooks/resharding.md): Runner recovery, durable respawn reset, safety checks, and local verification.
@@ -389,7 +391,7 @@ Run with config file:
 | `DUCKGRES_IDLE_TIMEOUT` | Connection idle timeout (e.g., `30m`, `1h`, `-1` to disable) | `24h` |
 | `DUCKGRES_CLIENT_IDLE_TIMEOUT_MAX` | Maximum client-requested `duckgres.idle_timeout`; unset disables client overrides | disabled |
 | `DUCKGRES_SESSION_INIT_TIMEOUT` | Session startup metadata initialization and catalog probe timeout | `10s` |
-| `DUCKGRES_WORKER_QUEUE_TIMEOUT` | Max time to wait for worker acquisition and per-org/per-user vCPU resource admission; the managed K8s queue TTL uses this value | `60s` |
+| `DUCKGRES_WORKER_QUEUE_TIMEOUT` | Max time to wait for worker acquisition and org memory/vCPU or user vCPU resource admission; the managed K8s queue TTL uses this value | `60s` |
 | `DUCKGRES_ADMISSION_RECLAIMER_MAX_RESERVATIONS` | Max queued/live admission identities whose cleanup ownership one control plane may retain; new admissions are rejected before enqueue when full | `4096` |
 | `DUCKGRES_HANDOVER_DRAIN_TIMEOUT` | Max time to drain planned shutdowns and upgrades before forcing exit | `24h` in process mode, `15m` in remote K8s mode |
 | `DUCKGRES_SNI_ROUTING_MODE` | Multi-tenant managed-hostname routing: `off`, `passthrough`, or `enforce`. Postgres uses the requested dbname first; managed SNI must resolve to the same org, and SNI supplies the database only when dbname is empty. | `off` |
@@ -839,7 +841,7 @@ Built-in rate limiting protects against brute-force authentication attacks:
 
 - **Failed attempt tracking**: Bans IPs after too many failed auth attempts
 - **Connection limits**: Limits concurrent connections per IP and, when configured, total concurrent sessions in standalone mode.
-- **K8s multi-tenant resource limits**: Org and user `max_vcpus` bound the sum of active worker pod vCPUs admitted through runtime-store leases. 0 means unlimited.
+- **K8s multi-tenant resource limits**: Org and user `max_vcpus` bound active admitted worker pod vCPUs, while org `max_memory` independently bounds active admitted worker pod memory. Zero means unlimited.
 - **Auto-cleanup**: Expired records are automatically cleaned up
 
 ```yaml
@@ -1034,8 +1036,8 @@ Managed-warehouse contract notes:
 - At most one managed-warehouse row exists per team. The row may be absent before first provisioning or after cleanup, but there is never more than one active warehouse contract for a team.
 - Each org has a `data_imports_table_naming_version`. Migration `000034` assigns `legacy_batch_v1` to orgs that already exist and changes the database default to `copy_v1` for orgs created afterward. `GET /api/v1/orgs/:id/teams` returns the org-level value alongside the team rows so every data-import reader and writer derives the same physical table name. Operators can change the policy in the admin console or with `PUT /api/v1/orgs/:id` using `{"data_imports_table_naming_version":"copy_v1"}`. Migrate existing tables before changing an org that has already written data.
 - The admin API exposes that contract at `GET /api/v1/teams/:name/warehouse` and `PUT /api/v1/teams/:name/warehouse`. Team list/get responses also include a nested `warehouse` object when present.
-- Org rows support optional `max_vcpus` on `POST /api/v1/orgs` and `PUT /api/v1/orgs/:id`. In K8s multi-tenant mode, this caps the org's active admitted worker pod vCPUs; `0` means unlimited.
-- Orgs automatically created during warehouse provisioning start with `max_vcpus=150`; `0` remains the explicit unlimited sentinel.
+- Org rows support optional `max_vcpus` and `max_memory` on `POST /api/v1/orgs` and `PUT /api/v1/orgs/:id`. In K8s multi-tenant mode these independently cap the org's active admitted worker pod vCPUs and memory. `max_memory` is a Kubernetes quantity such as `240Gi`; empty or `0` means unlimited.
+- Orgs automatically created during warehouse provisioning start with `max_vcpus=150` and unlimited memory. `0` remains the explicit unlimited sentinel.
 - User rows support an optional `max_vcpus` field on `POST /api/v1/users` and `PUT /api/v1/orgs/:id/users/:username`. `max_vcpus` limits the user's active admitted worker pod vCPUs in K8s multi-tenant mode; `0` means unlimited.
 - `PUT /api/v1/orgs/:id/teams/:team_id/project-reader` creates or rotates the generated SQL login for a PostHog project. The login can read every current and future table in the project's team, data-import, and modeled-data schemas, plus its legacy events/persons relations. Writes, unqualified application relations, external-reader and introspection functions, other projects' schemas, and their catalog metadata are denied by the PostgreSQL query gateway. The plaintext password is returned only by the rotation response.
 - The typed sections are `warehouse_database`, `metadata_store`, `s3`, `worker_identity`, and structured secret refs for `warehouse_database_credentials`, `metadata_store_credentials`, `s3_credentials`, and `runtime_config`. In shared worker mode, every non-empty secret ref must store an explicit `namespace`, and it must match `worker_identity.namespace`.

controlplane/admin/api.go

@@ -286,6 +286,7 @@ func (s *gormAPIStore) UpdateOrg(name string, updates configstore.Org) (*configs
 	fields := map[string]interface{}{
 		"max_workers": updates.MaxWorkers,
 		"max_vcpus":   updates.MaxVCPUs,
+		"max_memory":  updates.MaxMemory,
 		// Org default worker profile: written unconditionally so an explicit
 		// empty string CLEARS the default (the handler's presence-merge keeps
 		// omitted fields at their stored values before this runs).
@@ -1043,6 +1044,9 @@ func (h *apiHandler) updateOrg(c *gin.Context) {
 	if _, ok := fields["max_vcpus"]; ok {
 		merged.MaxVCPUs = updates.MaxVCPUs
 	}
+	if _, ok := fields["max_memory"]; ok {
+		merged.MaxMemory = updates.MaxMemory
+	}
 	// Org default worker profile: present-in-payload wins, including an
 	// explicit "" which clears the default.
 	if _, ok := fields["default_worker_cpu"]; ok {
@@ -1075,6 +1079,7 @@ func (h *apiHandler) updateOrg(c *gin.Context) {
 	addChange("database_name", existing.DatabaseName, merged.DatabaseName)
 	addChange("max_workers", existing.MaxWorkers, merged.MaxWorkers)
 	addChange("max_vcpus", existing.MaxVCPUs, merged.MaxVCPUs)
+	addChange("max_memory", orgStr(existing.MaxMemory), orgStr(merged.MaxMemory))
 	addChange("default_worker_cpu", orgStr(existing.DefaultWorkerCPU), orgStr(merged.DefaultWorkerCPU))
 	addChange("default_worker_memory", orgStr(existing.DefaultWorkerMemory), orgStr(merged.DefaultWorkerMemory))
 	addChange("default_worker_ttl", orgStr(existing.DefaultWorkerTTL), orgStr(merged.DefaultWorkerTTL))
@@ -1432,6 +1437,11 @@ func validateOrgMutationPayload(org *configstore.Org) error {
 	if org.MaxVCPUs < 0 {
 		return fmt.Errorf("max_vcpus: value %d must be >= 0", org.MaxVCPUs)
 	}
+	maxMemory, err := configstore.NormalizeOrgMaxMemory(org.MaxMemory)
+	if err != nil {
+		return fmt.Errorf("max_memory: %w", err)
+	}
+	org.MaxMemory = maxMemory
 	return nil
 }
 

controlplane/admin/api_test.go

@@ -81,6 +81,7 @@ func (s *fakeAPIStore) UpdateOrg(name string, updates configstore.Org) (*configs
 	}
 	org.MaxWorkers = updates.MaxWorkers
 	org.MaxVCPUs = updates.MaxVCPUs
+	org.MaxMemory = updates.MaxMemory
 	// Mirrors gormAPIStore: "" = preserve, non-empty renames.
 	if updates.DatabaseName != "" {
 		org.DatabaseName = updates.DatabaseName
@@ -2547,6 +2548,124 @@ func TestUpdateOrgRejectsNegativeMaxVCPUs(t *testing.T) {
 	}
 }
 
+func TestCreateOrgAcceptsMaxMemory(t *testing.T) {
+	store := newFakeAPIStore()
+	router := newTestAPIRouter(store)
+
+	rec := adminJSON(t, router, http.MethodPost, "/api/v1/orgs",
+		`{"name":"analytics","database_name":"analytics","team_id":1,"max_memory":"120Gi"}`)
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("status = %d, want %d: %s", rec.Code, http.StatusCreated, rec.Body.String())
+	}
+	if got := store.orgs["analytics"].MaxMemory; got != "120Gi" {
+		t.Fatalf("stored max_memory = %q, want 120Gi", got)
+	}
+	var response configstore.Org
+	if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if response.MaxMemory != "120Gi" {
+		t.Fatalf("response max_memory = %q, want 120Gi", response.MaxMemory)
+	}
+}
+
+func TestUpdateOrgMaxMemoryPersistsAndAudits(t *testing.T) {
+	store := newFakeAPIStore()
+	store.orgs["analytics"] = &configstore.Org{
+		Name: "analytics", MaxWorkers: 2, MaxMemory: "120Gi",
+	}
+
+	gin.SetMode(gin.TestMode)
+	router := gin.New()
+	var detail string
+	router.Use(func(c *gin.Context) {
+		c.Next()
+		detail = c.GetString(ctxAuditDetailKey)
+	})
+	registerAPIWithStore(router.Group("/api/v1"), store, nil, nil)
+
+	rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/analytics", `{"max_memory":"240Gi"}`)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d: %s", rec.Code, http.StatusOK, rec.Body.String())
+	}
+	if got := store.orgs["analytics"].MaxMemory; got != "240Gi" {
+		t.Fatalf("stored max_memory = %q, want 240Gi", got)
+	}
+	if got := store.orgs["analytics"].MaxWorkers; got != 2 {
+		t.Fatalf("max_workers = %d, want preserved 2", got)
+	}
+	if !strings.Contains(detail, "max_memory 120Gi → 240Gi") {
+		t.Fatalf("audit detail = %q, want max_memory change", detail)
+	}
+}
+
+func TestUpdateOrgOmittingMaxMemoryPreservesIt(t *testing.T) {
+	store := newFakeAPIStore()
+	store.orgs["analytics"] = &configstore.Org{
+		Name: "analytics", MaxWorkers: 2, MaxMemory: "120Gi",
+	}
+	router := newTestAPIRouter(store)
+
+	rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/analytics", `{"max_workers":3}`)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d: %s", rec.Code, http.StatusOK, rec.Body.String())
+	}
+	if got := store.orgs["analytics"].MaxMemory; got != "120Gi" {
+		t.Fatalf("max_memory = %q, want preserved 120Gi", got)
+	}
+}
+
+func TestUpdateOrgMaxMemoryCanBeCleared(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		body string
+	}{
+		{name: "empty string", body: `{"max_memory":""}`},
+		{name: "null", body: `{"max_memory":null}`},
+		{name: "zero quantity", body: `{"max_memory":"0"}`},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			store := newFakeAPIStore()
+			store.orgs["analytics"] = &configstore.Org{Name: "analytics", MaxMemory: "120Gi"}
+			router := newTestAPIRouter(store)
+
+			rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/analytics", tc.body)
+			if rec.Code != http.StatusOK {
+				t.Fatalf("status = %d, want %d: %s", rec.Code, http.StatusOK, rec.Body.String())
+			}
+			if got := store.orgs["analytics"].MaxMemory; got != "" {
+				t.Fatalf("max_memory = %q, want cleared", got)
+			}
+		})
+	}
+}
+
+func TestOrgMutationRejectsInvalidMaxMemory(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		body string
+	}{
+		{name: "invalid quantity", body: `{"max_memory":"lots"}`},
+		{name: "negative quantity", body: `{"max_memory":"-1Gi"}`},
+		{name: "decimal overflow", body: `{"max_memory":"10000000000000000000"}`},
+		{name: "binary overflow", body: `{"max_memory":"8Ei"}`},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			store := newFakeAPIStore()
+			store.orgs["analytics"] = &configstore.Org{Name: "analytics", MaxMemory: "120Gi"}
+			router := newTestAPIRouter(store)
+
+			rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/analytics", tc.body)
+			if rec.Code != http.StatusBadRequest {
+				t.Fatalf("status = %d, want %d: %s", rec.Code, http.StatusBadRequest, rec.Body.String())
+			}
+			if got := store.orgs["analytics"].MaxMemory; got != "120Gi" {
+				t.Fatalf("invalid update changed max_memory to %q", got)
+			}
+		})
+	}
+}
+
 func TestUpdateOrgDataImportsTableNamingVersion(t *testing.T) {
 	for _, tc := range []struct {
 		name string

controlplane/admin/embed_ui_test.go

@@ -5,6 +5,7 @@ package admin
 import (
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"testing"
 
@@ -49,3 +50,19 @@ func TestRegisterUIServesSPA(t *testing.T) {
 		})
 	}
 }
+
+func TestModelsUIExplainsAndEditsOrgMaxMemory(t *testing.T) {
+	raw, err := os.ReadFile("static/models.html")
+	if err != nil {
+		t.Fatalf("read models UI: %v", err)
+	}
+	body := string(raw)
+	for _, want := range []string{
+		`max_memory: "Maximum active worker pod memory`,
+		`max_memory: "text"`,
+	} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("models UI missing %q", want)
+		}
+	}
+}

controlplane/admin/monitoring.go

@@ -45,6 +45,7 @@ type monitoringWarehouse struct {
 type monitoringLimits struct {
 	MaxWorkers              int    `json:"max_workers"`
 	MaxVCPUs                int    `json:"max_vcpus"`
+	MaxMemoryBytes          int64  `json:"max_memory_bytes"`
 	DefaultWorkerCPU        string `json:"default_worker_cpu"`
 	DefaultWorkerMemory     string `json:"default_worker_memory"`
 	DefaultWorkerTTLSeconds int    `json:"default_worker_ttl_seconds"`
@@ -292,6 +293,7 @@ func (h *monitoringHandler) snapshot(c *gin.Context) {
 		Limits: monitoringLimits{
 			MaxWorkers:              org.MaxWorkers,
 			MaxVCPUs:                org.MaxVCPUs,
+			MaxMemoryBytes:          org.MaxMemoryBytes,
 			DefaultWorkerCPU:        configuredDefaults.CPU,
 			DefaultWorkerMemory:     configuredDefaults.Memory,
 			DefaultWorkerTTLSeconds: int(configuredDefaults.TTL / time.Second),

controlplane/admin/monitoring_test.go

@@ -57,6 +57,7 @@ func TestMonitoringSnapshotIsOrgScopedSanitizedAndReportsPartialCoverage(t *test
 				Name:                    "org-a",
 				MaxWorkers:              4,
 				MaxVCPUs:                8,
+				MaxMemoryBytes:          16 * 1024 * 1024 * 1024,
 				DefaultWorkerCPU:        "1",
 				DefaultWorkerMemory:     "2Gi",
 				DefaultWorkerTTL:        "30m",
@@ -131,6 +132,9 @@ func TestMonitoringSnapshotIsOrgScopedSanitizedAndReportsPartialCoverage(t *test
 	if got.Limits.DefaultWorkerCPU != "1" || got.Limits.DefaultWorkerMemory != "2Gi" || got.Limits.DefaultWorkerTTLSeconds != 1800 {
 		t.Fatalf("configured defaults = %+v, want current org defaults", got.Limits)
 	}
+	if got.Limits.MaxMemoryBytes != 16*1024*1024*1024 {
+		t.Fatalf("max memory bytes = %d, want 16Gi", got.Limits.MaxMemoryBytes)
+	}
 	if got.Totals.Workers != 2 || got.Totals.AllocatedCPUCores != 2.75 || got.Totals.AllocatedMemoryBytes != 7*1024*1024*1024 {
 		t.Fatalf("worker totals = %+v, want 2 workers / 2.75 cores / 7Gi", got.Totals)
 	}

controlplane/admin/static/models.html

@@ -366,6 +366,7 @@ <h2>Session Expired</h2>
       database_name: "Database name clients connect with (dbname=). Unique across orgs.",
       max_workers: "Maximum concurrent worker pods this org may run. 0 means use the deployment default.",
       max_vcpus: "Maximum active worker pod vCPUs this org may have admitted. 0 means unlimited.",
+      max_memory: "Maximum active worker pod memory this org may have admitted, as a Kubernetes quantity. Empty means unlimited.",
       default_worker_cpu: "Operator-set default worker pod CPU (k8s quantity, e.g. 2). Empty means unset. Applied when a connection does not size itself via duckgres.worker_* options.",
       default_worker_memory: "Operator-set default worker pod memory (k8s quantity, e.g. 8Gi). Empty means unset.",
       default_worker_ttl: "Operator-set default hot-idle TTL for workers (Go duration string, e.g. 75m). Empty means unset.",
@@ -453,6 +454,7 @@ <h2>Session Expired</h2>
       pid: "Client backend PID associated with the request.",
       protocol: "Session interface for the request (postgres or admin).",
       requested_vcpus: "Worker pod vCPUs requested by this admission entry.",
+      requested_memory_bytes: "Worker pod memory bytes requested by this admission entry.",
       enqueued_at: "When the request joined the FIFO queue; orders fairness.",
       expires_at: "When this queue entry expires and is reclaimed.",
       granted_at: "When admission was granted, if applicable.",
@@ -468,7 +470,8 @@ <h2>Session Expired</h2>
       pid: "Client backend PID holding the lease.",
       protocol: "Session interface for the lease (postgres or admin).",
       requested_vcpus: "Worker pod vCPUs charged to the org and user budgets.",
-      acquired_at: "When the lease was acquired; counts toward active vCPU resource budgets.",
+      requested_memory_bytes: "Worker pod memory bytes charged to the org budget.",
+      acquired_at: "When the lease was acquired; counts toward active vCPU and memory resource budgets.",
       created_at: "When this lease record was created.",
       updated_at: "When this lease record was last modified.",
     },
@@ -483,8 +486,8 @@ <h2>Session Expired</h2>
     "managed-warehouses": "The config-store source of truth for an org's managed warehouse: image, metadata/object stores, catalogs, identity, and provisioning status.",
     "cp-instances": "A runtime coordination record for one control-plane process, with liveness state and heartbeat.",
     "worker-records": "A durable runtime record for one worker pod: lifecycle state, owner, profile, TTL, and credential expiry.",
-    "org-connection-queue": "A cluster-wide FIFO resource admission request for one org/user session; rows expire quickly and coordinate fairness across CP replicas.",
-    "org-connection-leases": "A durable cluster-wide resource admission lease for a live session; active leases count toward org and user vCPU budgets.",
+    "org-connection-queue": "A cluster-wide FIFO CPU and memory admission request for one org/user session; rows expire quickly and coordinate fairness across CP replicas.",
+    "org-connection-leases": "A durable cluster-wide resource admission lease for a live session; active leases count toward org memory and org/user vCPU budgets.",
   };
 
   // Per-org aux data computed client-side: the live /models/orgs listing does
@@ -715,6 +718,7 @@ <h2>Session Expired</h2>
   const ORG_EDIT_FIELDS = {
     max_workers: "int",
     max_vcpus: "int",
+    max_memory: "text",
     default_worker_cpu: "text",
     default_worker_memory: "text",
     default_worker_ttl: "text",