feat: register Trino cells and select initial warehouse placement

CLAUDE.md

@@ -1551,8 +1551,10 @@ limits on a shared multi-tenant Trino cluster called a **cell**. The control
 plane is the only writer of that state: `provisioner/trino_provisioner.go`
 projects it every controller tick from `duckgres_managed_warehouse_trino` +
 the org's warehouse row + its Duckling CR. Enablement is env-inferred —
-`DUCKGRES_TRINO_COORDINATOR_URL` set means on (`controlplane/trino_inputs.go`);
-unset means the branch never wires and nothing changes. **Trino is binary: if
+`DUCKGRES_TRINO_COORDINATOR_URL` identifies legacy (`controlplane/trino_inputs.go`).
+The optional `DUCKGRES_TRINO_CELLS_FILE` adds namespace-isolated logical cells
+with blue/green backends; it still requires legacy configuration. With neither
+setting, the branch never wires and nothing changes. **Trino is binary: if
 you asked for it, a wiring failure is fatal at startup**, because silently
 skipping leaves the cell's OPA sidecar serving a last-good bundle while
 password/tenant/catalog changes never propagate.
@@ -1674,12 +1676,22 @@ password/tenant/catalog changes never propagate.
   breaks every reconcile tick's own DDL.
 - **Cells, minimally**: `trino_cell_id` on the row names the owning cell
   (`DUCKGRES_TRINO_CELL_ID`, default `configstore.DefaultTrinoCellID`). A
-  provisioner claims unassigned orgs (`AssignTrinoCell`, conditional in SQL so
+  provisioner claims unassigned orgs (`ClaimTrinoCell`, conditional in SQL so
   no cell can steal another's tenant), reconciles its own, and ignores the
-  rest — including writing NO state for them. There is exactly ONE cell today
-  and deliberately no assignment policy, capacity model, rebalancer or cell
-  drain; `resolveTrinoCell` becoming `resolveTrinoCells` is the whole shape of
-  adding a second.
+  rest — including writing NO state for them. Claims return an authoritative
+  boolean: a lost claim must never project the losing cell's tenant. Only
+  legacy claims unassigned warehouses. Registered logical IDs have the reserved
+  storage prefix `registered:`; the old stored `cell-001` remains legacy.
+  Admin-only initial selection runs before first enablement and refuses changes
+  to any already owned warehouse, including a disabled one. No maintenance move,
+  capacity model, rebalancer, drain, or Gateway routing controller is included.
+  See [docs/trino-cells.md](docs/trino-cells.md) for configuration and recovery.
+- **Blue/green projections share one logical cell's namespace**, but internal
+  communication Secrets remain distinct, chart-owned read-only references.
+  Stopped backends do not receive catalog calls or observer polling. Every
+  running backend must reconcile its independent catalog set before the org is
+  ready. Cell failures do not block sibling-cell projections. Configuration is
+  startup-loaded and must agree across control-plane replicas.
 - **The existing deployment's API identity is `legacy`.** The Trino console
   exposes this name in `cell.id` and in owned orgs' `status.cell` / `orgs[].cell`.
   `TrinoCell.StoredID` keeps the configured ownership ID private to the adapter.
@@ -1690,7 +1702,8 @@ password/tenant/catalog changes never propagate.
 - **The bundle endpoint is mounted OUTSIDE `/api/v1`** (`/bundles/trino`) with
   its own bearer auth, and `buildTrinoWiring` bootstraps SYNCHRONOUSLY so the
   handler is constructed with the real token — there is no window where it
-  serves under a placeholder.
+  serves under a placeholder. Registered cells use `/bundles/trino/<cell-id>`;
+  each cell's token can read only its own bundle. Legacy retains the old path.
 - Touching any of this → update `provisioner/trino_provisioner_test.go`,
   `provisioner/trino_cluster_secrets_test.go`, `provisioner/opa/*_test.go`,
   `provisioning/api_test.go`, `tests/configstore/trino_postgres_test.go` +

README.md

@@ -1,5 +1,8 @@
 # Duckgres
 
+Trino operators: see [cell registration and placement](docs/trino-cells.md) for
+the optional registry, unchanged legacy defaults, and migration limitations.
+
 <p align="center">
   <img src="media/oh_duck.png" alt="Duckgres Mascot" width="200">
 </p>

controlplane/admin/README.md

@@ -194,9 +194,26 @@ The existing deployment appears as `legacy` in the Trino API's `cell.id` and
 its owned orgs' `status.cell` / `orgs[].cell`. This is an API alias, not a storage
 migration: `DUCKGRES_TRINO_CELL_ID`, persisted org assignments, and Trino's
 catalog-store key retain their existing values. The general org endpoint still
-returns the persisted `trino.trino_cell_id`. Unassigned and foreign-cell rows
-retain their original IDs and receive no connection details. This change adds
-no cell selection or tenant migration endpoint.
+returns the persisted `trino.trino_cell_id`. Registered cells use separate
+`registered:<cell-id>` ownership values. Unknown stored owners fail closed.
+
+`GET /api/v1/trino/cells` lists configured logical cells. Operational Trino
+routes accept `?cell=<logical-id>` and default to `legacy`. Org detail resolves
+its authoritative stored assignment regardless of a supplied cell parameter.
+Each coordinator has separate caches; tenant counts include only its cell.
+
+Admins can select an initial cell in the org's Trino card or send
+`PUT /api/v1/orgs/:id/trino/cell` with `{"cell":"cell-001"}`. This stores a
+disabled assignment before the normal enable endpoint is called. The warehouse
+must already exist. Selection does not enable Trino or move an existing tenant.
+An existing assignment is immutable, including after disable/re-enable; repeat
+selection of the same assignment is idempotent. An enabled unassigned row also
+rejects selection because its first provisioning tick may already be running.
+On a 409, inspect the current assignment rather than editing its database row.
+Moving an existing tenant requires a separate maintenance/drain workflow.
+
+For newly registered cells, configure the shared customer endpoint separately
+from the observer coordinator URL. This API does not configure Gateway routing.
 
 For local verification, run `just test-controlplane-k8s` and `just ui-test`.
 The isolated Trino end-to-end suite checks both the API alias and unchanged

controlplane/admin/trino.go

@@ -225,13 +225,15 @@ func (c *trinoCache[T]) get(ctx context.Context, fetch func(context.Context) (T,
 // TrinoAPI is the console's Trino surface: a coordinator client, the config
 // store, and the cell's identity.
 type TrinoAPI struct {
-	cell    TrinoCell
-	client  TrinoCoordinatorClient
-	orgs    TrinoOrgStore
-	audit   *AuditStore
-	queries *trinoCache[[]TrinoQuery]
-	nodes   *trinoCache[TrinoNodeInventory]
-	info    *trinoCache[*TrinoServerInfo]
+	fleet      map[string]*TrinoAPI
+	filterCell bool
+	cell       TrinoCell
+	client     TrinoCoordinatorClient
+	orgs       TrinoOrgStore
+	audit      *AuditStore
+	queries    *trinoCache[[]TrinoQuery]
+	nodes      *trinoCache[TrinoNodeInventory]
+	info       *trinoCache[*TrinoServerInfo]
 }
 
 // NewTrinoAPI builds the console's Trino surface. A nil client or store
@@ -259,6 +261,10 @@ func registerTrinoAPI(r *gin.RouterGroup, api *TrinoAPI) {
 	if api == nil {
 		return
 	}
+	if api.fleet != nil {
+		registerTrinoFleetAPI(r, api)
+		return
+	}
 	r.GET("/trino/status", api.handleStatus)
 	r.GET("/trino/queries", api.handleQueries)
 	r.GET("/trino/queries/:id", api.handleQueryDetail)
@@ -281,7 +287,15 @@ func (a *TrinoAPI) index() (principalIndex, error) {
 		return principalIndex{}, err
 	}
 	idx := principalIndex{orgByPrincipal: make(map[string]string, len(rows)), rows: rows}
-	for _, o := range rows {
+	if a.filterCell {
+		idx.rows = nil
+		for _, row := range rows {
+			if row.CellID == a.cell.storedID() || (row.CellID == "" && a.cell.ID == "legacy") {
+				idx.rows = append(idx.rows, row)
+			}
+		}
+	}
+	for _, o := range idx.rows {
 		if p := o.TrinoPrincipal(); p != "" {
 			idx.orgByPrincipal[p] = o.OrgID
 		}
@@ -623,10 +637,14 @@ func (a *TrinoAPI) handleOrgDetail(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
+	a.writeOrgDetail(c, orgID, row)
+}
+
+func (a *TrinoAPI) writeOrgDetail(c *gin.Context, orgID string, row *configstore.ManagedWarehouseTrino) {
 	if row == nil || !row.Enabled {
 		// Not an error: most orgs are not Trino-enabled, and the org page
 		// renders a "not enabled" state rather than a failure.
-		c.JSON(http.StatusOK, gin.H{"cell": a.cell, "enabled": false})
+		c.JSON(http.StatusOK, gin.H{"cell": a.cell, "enabled": false, "assigned": row != nil && row.TrinoCellID != ""})
 		return
 	}
 
@@ -677,6 +695,7 @@ func (a *TrinoAPI) handleOrgDetail(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{
 		"cell":      a.cell,
 		"enabled":   true,
+		"assigned":  row.TrinoCellID != "",
 		"available": available,
 		"status":    status,
 	})

controlplane/admin/trino_fleet.go

@@ -0,0 +1,126 @@
+//go:build kubernetes
+
+package admin
+
+import (
+	"errors"
+	"net/http"
+	"sort"
+
+	"github.com/gin-gonic/gin"
+	"github.com/posthog/duckgres/controlplane/configstore"
+)
+
+// NewTrinoFleetAPI builds isolated coordinator views and preserves the legacy default.
+func NewTrinoFleetAPI(cells []TrinoCell, clients []TrinoCoordinatorClient, orgs TrinoOrgStore, audit *AuditStore) *TrinoAPI {
+	if len(cells) == 0 || len(cells) != len(clients) || orgs == nil {
+		return nil
+	}
+	a := &TrinoAPI{fleet: make(map[string]*TrinoAPI), orgs: orgs, audit: audit}
+	stored := make(map[string]bool)
+	for i, cell := range cells {
+		if cell.ID == "" || a.fleet[cell.ID] != nil || stored[cell.storedID()] {
+			return nil
+		}
+		child := NewTrinoAPI(cell, clients[i], orgs, audit)
+		if child == nil {
+			return nil
+		}
+		child.filterCell = true
+		a.fleet[cell.ID] = child
+		stored[cell.storedID()] = true
+	}
+	return a
+}
+
+func registerTrinoFleetAPI(r *gin.RouterGroup, api *TrinoAPI) {
+	r.GET("/trino/cells", api.handleCells)
+	r.GET("/trino/status", api.forCell((*TrinoAPI).handleStatus))
+	r.GET("/trino/queries", api.forCell((*TrinoAPI).handleQueries))
+	r.GET("/trino/queries/:id", api.forCell((*TrinoAPI).handleQueryDetail))
+	r.POST("/trino/queries/:id/kill", api.forCell((*TrinoAPI).handleKillQuery))
+	r.GET("/trino/nodes", api.forCell((*TrinoAPI).handleNodes))
+	r.GET("/trino/orgs", api.forCell((*TrinoAPI).handleOrgs))
+	r.GET("/orgs/:id/trino", api.handleFleetOrg)
+	r.PUT("/orgs/:id/trino/cell", api.handleSelectCell)
+}
+
+func (a *TrinoAPI) forCell(handle func(*TrinoAPI, *gin.Context)) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		id := c.DefaultQuery("cell", "legacy")
+		selected := a.fleet[id]
+		if selected == nil {
+			c.JSON(http.StatusNotFound, gin.H{"error": "unknown Trino cell"})
+			return
+		}
+		handle(selected, c)
+	}
+}
+
+func (a *TrinoAPI) handleCells(c *gin.Context) {
+	cells := make([]TrinoCell, 0, len(a.fleet))
+	for _, cell := range a.fleet {
+		cells = append(cells, cell.cell)
+	}
+	sort.Slice(cells, func(i, j int) bool { return cells[i].ID < cells[j].ID })
+	c.JSON(http.StatusOK, gin.H{"cells": cells})
+}
+
+func (a *TrinoAPI) handleFleetOrg(c *gin.Context) {
+	row, err := a.orgs.GetManagedWarehouseTrino(c.Param("id"))
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "cannot read Trino assignment"})
+		return
+	}
+	selected := a.fleet["legacy"]
+	if row != nil && row.TrinoCellID != "" {
+		selected = nil
+		for _, candidate := range a.fleet {
+			if candidate.cell.storedID() == row.TrinoCellID {
+				selected = candidate
+				break
+			}
+		}
+	}
+	if selected == nil {
+		c.JSON(http.StatusConflict, gin.H{"error": "the assigned Trino cell is not configured", "assigned": row != nil && row.TrinoCellID != ""})
+		return
+	}
+	selected.writeOrgDetail(c, c.Param("id"), row)
+}
+
+type trinoCellSelector interface {
+	SelectTrinoCell(orgID, cellID string) error
+}
+
+func (a *TrinoAPI) handleSelectCell(c *gin.Context) {
+	identity := IdentityFromContext(c)
+	if identity == nil || identity.Role != RoleAdmin {
+		c.JSON(http.StatusForbidden, gin.H{"error": "admin role required"})
+		return
+	}
+	var req struct {
+		Cell string `json:"cell" binding:"required"`
+	}
+	if err := c.ShouldBindJSON(&req); err != nil || a.fleet[req.Cell] == nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "a configured Trino cell is required"})
+		return
+	}
+	selector, ok := a.orgs.(trinoCellSelector)
+	if !ok {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Trino cell selection is unavailable"})
+		return
+	}
+	if err := selector.SelectTrinoCell(c.Param("id"), a.fleet[req.Cell].cell.storedID()); err != nil {
+		switch {
+		case errors.Is(err, configstore.ErrTrinoCellSelectionConflict):
+			c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
+		case errors.Is(err, configstore.ErrTrinoWarehouseNotFound):
+			c.JSON(http.StatusNotFound, gin.H{"error": err.Error()})
+		default:
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "cannot select Trino cell"})
+		}
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"cell": a.fleet[req.Cell].cell, "assigned": true})
+}

controlplane/admin/trino_fleet_test.go

@@ -0,0 +1,95 @@
+//go:build kubernetes
+
+package admin
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/posthog/duckgres/controlplane/configstore"
+)
+
+func TestTrinoFleetSelectsAuthoritativeOwner(t *testing.T) {
+	legacy := &fakeTrinoCoordinator{}
+	current := &fakeTrinoCoordinator{queries: []TrinoQuery{{Principal: "tenant"}}}
+	store := &fakeTrinoOrgStore{
+		orgs: []configstore.TrinoEnabledOrg{{OrgID: "tenant", DatabaseName: "tenant", CellID: "registered:cell-001"}},
+		rows: map[string]*configstore.ManagedWarehouseTrino{"tenant": {OrgID: "tenant", Enabled: true, TrinoCellID: "registered:cell-001", State: configstore.ManagedWarehouseStateReady}},
+	}
+	api := NewTrinoFleetAPI([]TrinoCell{{ID: "legacy", StoredID: "cell-001"}, {ID: "cell-001", StoredID: "registered:cell-001", ClientURL: "https://gateway.example.com"}}, []TrinoCoordinatorClient{legacy, current}, store, nil)
+	r := trinoTestRouter(api, RoleAdmin)
+	code, data := doTrinoJSON(t, r, http.MethodGet, "/api/v1/trino/queries?cell=cell-001", "")
+	if code != http.StatusOK || data["cell"].(map[string]any)["id"] != "cell-001" || legacy.queryCalls.Load() != 0 || current.queryCalls.Load() != 1 {
+		t.Fatalf("wrong coordinator: %d %#v", code, data)
+	}
+	code, data = doTrinoJSON(t, r, http.MethodGet, "/api/v1/orgs/tenant/trino?cell=legacy", "")
+	if code != http.StatusOK || data["cell"].(map[string]any)["id"] != "cell-001" {
+		t.Fatalf("org must resolve its stored owner: %d %#v", code, data)
+	}
+	status := data["status"].(map[string]any)
+	if status["connection"].(map[string]any)["host"] != "gateway.example.com" {
+		t.Fatalf("wrong connection: %#v", status)
+	}
+	for _, cell := range []string{"legacy", "cell-001"} {
+		code, data = doTrinoJSON(t, r, http.MethodGet, "/api/v1/trino/orgs?cell="+cell, "")
+		want := 0
+		if cell == "cell-001" {
+			want = 1
+		}
+		if code != http.StatusOK || len(data["orgs"].([]any)) != want {
+			t.Fatalf("org listing crossed cell boundary: %d %#v", code, data)
+		}
+	}
+	for _, path := range []string{"/api/v1/trino/queries", "/api/v1/trino/status", "/api/v1/trino/nodes", "/api/v1/trino/orgs", "/api/v1/trino/queries/query"} {
+		code, _ = doTrinoJSON(t, r, http.MethodGet, path+"?cell=unknown", "")
+		if code != http.StatusNotFound {
+			t.Errorf("%s unknown cell returned %d", path, code)
+		}
+	}
+	store.rows["tenant"].TrinoCellID = "unregistered"
+	code, data = doTrinoJSON(t, r, http.MethodGet, "/api/v1/orgs/tenant/trino", "")
+	if code != http.StatusConflict || data["status"] != nil {
+		t.Fatalf("unknown stored owner must fail closed: %d %#v", code, data)
+	}
+}
+
+func TestTrinoFleetRejectsAmbiguousConfiguration(t *testing.T) {
+	for _, cells := range [][]TrinoCell{
+		{{ID: "legacy", StoredID: "same"}, {ID: "cell-001", StoredID: "same"}},
+		{{ID: "legacy", StoredID: "first"}, {ID: "legacy", StoredID: "second"}},
+		{{ID: ""}, {ID: "cell-001"}},
+	} {
+		if api := NewTrinoFleetAPI(cells, []TrinoCoordinatorClient{&fakeTrinoCoordinator{}, &fakeTrinoCoordinator{}}, &fakeTrinoOrgStore{}, nil); api != nil {
+			t.Fatalf("ambiguous fleet accepted: %+v", cells)
+		}
+	}
+}
+
+type selectingTrinoStore struct {
+	fakeTrinoOrgStore
+	selected string
+}
+
+func (s *selectingTrinoStore) SelectTrinoCell(_, cell string) error {
+	s.selected = cell
+	return nil
+}
+
+func TestTrinoFleetSelectionIsAdminOnly(t *testing.T) {
+	store := &selectingTrinoStore{}
+	api := NewTrinoFleetAPI([]TrinoCell{{ID: "legacy", StoredID: "cell-001"}, {ID: "cell-001", StoredID: "registered:cell-001"}}, []TrinoCoordinatorClient{&fakeTrinoCoordinator{}, &fakeTrinoCoordinator{}}, store, nil)
+	for _, role := range []Role{RoleViewer, RoleAdmin} {
+		code, _ := doTrinoJSON(t, trinoTestRouter(api, role), http.MethodPut, "/api/v1/orgs/tenant/trino/cell", `{"cell":"cell-001"}`)
+		if role == RoleViewer && (code != http.StatusForbidden || store.selected != "") {
+			t.Fatalf("viewer changed assignment: %d", code)
+		}
+		if role == RoleAdmin && (code != http.StatusOK || store.selected != "registered:cell-001") {
+			t.Fatalf("admin assignment: %d %q", code, store.selected)
+		}
+	}
+	store.selected = ""
+	code, _ := doTrinoJSON(t, trinoTestRouter(api, RoleAdmin), http.MethodPut, "/api/v1/orgs/tenant/trino/cell", `{"cell":"unknown"}`)
+	if code != http.StatusBadRequest || store.selected != "" {
+		t.Fatalf("unknown cell selected: %d", code)
+	}
+}