feat: replace pooled Trino instances before voluntary node disruption

CLAUDE.md

@@ -1912,6 +1912,24 @@ the Trino backend selection).
   below compares the two).
 - **The pool keeps the cell's identity**: routing group and namespace are
   unchanged. Replacing compute never rewrites which warehouse lives where.
+- **Voluntary node replacement is opt-in** through
+  `DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED` (default false, requires the pool
+  operator). Charts protect new coordinator and worker templates with
+  `karpenter.sh/do-not-disrupt=true`. The enabled operator protects existing pod
+  metadata after checking pod/ReplicaSet/recorded Deployment UID ownership.
+  A complete NodeClaim inventory and exact node/provider/claim identity establish
+  Drifted or deleting-node evidence. Migration 000043 preserves the first request
+  per instance under the pool fence. The operator then cordons that exact node
+  with a UID/resource-version update; it never evicts pods or deletes nodes.
+  This is not SIGTERM-triggered draining. Normal desired+surge capacity admits a
+  replacement before draining the original, even when desired exceeds the floor.
+  No repair allowance or query timeout is added. Current Gateway obligations and
+  UID-scoped retirement still control deletion. Fresh candidates reject unsafe
+  nodes; ambiguous admission first replays its original Gateway intent.
+  Node lookup errors block voluntary rollout, not health checks, explicit
+  recovery, or already-authorized retirement. Disabling observation does not
+  cancel a durable replacement request. See the
+  [node replacement runbook](docs/runbooks/trino-pool-node-replacement.md).
 - **The catalog store's partition is its OWN identity**
   (`DUCKGRES_TRINO_POOL_CATALOG_CELL_ID`, resolved by
   `trinoPoolCatalogCellID`): it is the value every coordinator of the cell

README.md

@@ -3,6 +3,11 @@
 Trino operators: see [cell registration and placement](docs/trino-cells.md) for
 the optional registry, unchanged legacy defaults, and migration limitations.
 
+`DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED` defaults to `false`. Enabled shared-pool
+operators protect owned pods and replace instances on drifted or deleting Karpenter
+nodes through normal admission and drain. See the [node replacement runbook](docs/runbooks/trino-pool-node-replacement.md)
+for required permissions, rollout order, and failure boundaries.
+
 <p align="center">
   <img src="media/oh_duck.png" alt="Duckgres Mascot" width="200">
 </p>

controlplane/admin/trino_recovery.go

@@ -38,10 +38,11 @@ type trinoRecoveryRequest struct {
 }
 
 type trinoRecoveryInstanceSummary struct {
-	InstanceID     string    `json:"instance_id"`
-	Phase          string    `json:"phase"`
-	GatewayState   string    `json:"gateway_state"`
-	PhaseChangedAt time.Time `json:"phase_changed_at"`
+	InstanceID      string                             `json:"instance_id"`
+	Phase           string                             `json:"phase"`
+	GatewayState    string                             `json:"gateway_state"`
+	PhaseChangedAt  time.Time                          `json:"phase_changed_at"`
+	NodeReplacement *trinopool.NodeReplacementEvidence `json:"node_replacement,omitempty"`
 }
 
 func (a *TrinoAPI) handleRecoveryInstances(c *gin.Context) {
@@ -75,10 +76,17 @@ func (a *TrinoAPI) handleRecoveryInstances(c *gin.Context) {
 		if instance.PoolID != pool.PoolID || trinopool.Phase(instance.Phase).Terminal() {
 			continue
 		}
-		rows = append(rows, trinoRecoveryInstanceSummary{
+		row := trinoRecoveryInstanceSummary{
 			InstanceID: instance.InstanceID, Phase: instance.Phase,
 			GatewayState: instance.GatewayState, PhaseChangedAt: instance.PhaseChangedAt,
-		})
+		}
+		if instance.NodeReplacementEvidence != nil {
+			var evidence trinopool.NodeReplacementEvidence
+			if json.Unmarshal([]byte(*instance.NodeReplacementEvidence), &evidence) == nil && evidence.Validate() == nil {
+				row.NodeReplacement = &evidence
+			}
+		}
+		rows = append(rows, row)
 	}
 	sort.Slice(rows, func(i, j int) bool { return rows[i].InstanceID < rows[j].InstanceID })
 	c.JSON(http.StatusOK, gin.H{"cell": a.cell.ID, "instances": rows})

controlplane/admin/trino_recovery_test.go

@@ -94,6 +94,50 @@ func TestTrinoRecoveryInstancesAreScopedReadOnlyAndSanitized(t *testing.T) {
 	}
 }
 
+func TestTrinoRecoveryInstancesExposeTypedNodeReplacementEvidence(t *testing.T) {
+	api, store := recoveryTrinoAPI()
+	for _, test := range []struct {
+		name, evidence string
+		present        bool
+	}{
+		{"valid", `{"node_name":"node-a","node_uid":"node-uid-a","nodeclaim_name":"claim-a","nodeclaim_uid":"claim-uid-a","reason":"Drifted","private_extra":"do-not-expose"}`, true},
+		{"malformed", `not-json`, false},
+		{"incomplete", `{"reason":"Drifted"}`, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			instance := *store.instance
+			instance.NodeReplacementEvidence = &test.evidence
+			store.instances = []configstore.TrinoPoolInstance{instance}
+			code, data := doTrinoJSON(t, trinoTestRouter(api, RoleAdmin), http.MethodGet, "/api/v1/trino/instances?cell=pool-a", "")
+			if code != http.StatusOK || store.writes != 0 {
+				t.Fatalf("inventory: %d %#v", code, data)
+			}
+			row := data["instances"].([]any)[0].(map[string]any)
+			projection, ok := row["node_replacement"].(map[string]any)
+			if ok != test.present {
+				t.Fatalf("unexpected evidence projection: %#v", row)
+			}
+			if ok {
+				want := map[string]string{"node_name": "node-a", "node_uid": "node-uid-a", "nodeclaim_name": "claim-a", "nodeclaim_uid": "claim-uid-a", "reason": "Drifted"}
+				if len(projection) != len(want) || len(row) != 5 {
+					t.Fatalf("unexpected fields: %#v", row)
+				}
+				for key, value := range want {
+					if projection[key] != value {
+						t.Fatalf("%s = %#v", key, projection[key])
+					}
+				}
+			}
+			encoded, _ := json.Marshal(data)
+			for _, forbidden := range []string{"do-not-expose", "private.example.test", "blueprint", "endpoint_url", "private_extra"} {
+				if strings.Contains(string(encoded), forbidden) {
+					t.Fatalf("exposes %s", forbidden)
+				}
+			}
+		})
+	}
+}
+
 func TestTrinoRecoveryInstancesRejectViewerAndUnavailablePool(t *testing.T) {
 	api, store := recoveryTrinoAPI()
 	path := "/api/v1/trino/instances?cell=pool-a"

controlplane/configstore/migrations/000043_add_trino_node_replacement.sql

@@ -0,0 +1,7 @@
+-- +goose Up
+ALTER TABLE duckgres_trino_pool_instances
+    ADD COLUMN node_replacement_evidence JSONB;
+
+-- +goose Down
+ALTER TABLE duckgres_trino_pool_instances
+    DROP COLUMN node_replacement_evidence;

controlplane/configstore/trino_pool_models.go

@@ -129,20 +129,21 @@ type TrinoPoolInstance struct {
 	// RepairFor names the failed instance this one replaces. The Gateway
 	// charges an activation to the repair budget only when it is set; without
 	// it a repair spends the single planned surge instead.
-	RepairFor                 string `gorm:"column:repair_for"`
-	FailureReason             string `gorm:"column:failure_reason"`
-	CoordinatorDeploymentName string `gorm:"column:coordinator_deployment_name"`
-	CoordinatorDeploymentUID  string `gorm:"column:coordinator_deployment_uid"`
-	WorkerDeploymentName      string `gorm:"column:worker_deployment_name"`
-	WorkerDeploymentUID       string `gorm:"column:worker_deployment_uid"`
-	ServiceName               string `gorm:"column:service_name"`
-	ServiceUID                string `gorm:"column:service_uid"`
-	ConfigMapName             string `gorm:"column:config_map_name"`
-	ConfigMapUID              string `gorm:"column:config_map_uid"`
-	WorkerConfigMapName       string `gorm:"column:worker_config_map_name"`
-	WorkerConfigMapUID        string `gorm:"column:worker_config_map_uid"`
-	CoordinatorPodUID         string `gorm:"column:coordinator_pod_uid"`
-	CoordinatorNodeID         string `gorm:"column:coordinator_node_id"`
+	RepairFor                 string  `gorm:"column:repair_for"`
+	NodeReplacementEvidence   *string `gorm:"column:node_replacement_evidence;type:jsonb"`
+	FailureReason             string  `gorm:"column:failure_reason"`
+	CoordinatorDeploymentName string  `gorm:"column:coordinator_deployment_name"`
+	CoordinatorDeploymentUID  string  `gorm:"column:coordinator_deployment_uid"`
+	WorkerDeploymentName      string  `gorm:"column:worker_deployment_name"`
+	WorkerDeploymentUID       string  `gorm:"column:worker_deployment_uid"`
+	ServiceName               string  `gorm:"column:service_name"`
+	ServiceUID                string  `gorm:"column:service_uid"`
+	ConfigMapName             string  `gorm:"column:config_map_name"`
+	ConfigMapUID              string  `gorm:"column:config_map_uid"`
+	WorkerConfigMapName       string  `gorm:"column:worker_config_map_name"`
+	WorkerConfigMapUID        string  `gorm:"column:worker_config_map_uid"`
+	CoordinatorPodUID         string  `gorm:"column:coordinator_pod_uid"`
+	CoordinatorNodeID         string  `gorm:"column:coordinator_node_id"`
 	// CoordinatorID is the coordinator identity the GATEWAY observed when the
 	// member registered. It is a distinct value from the node id, and a loss
 	// claim has to carry both exactly as the Gateway recorded them, or the
@@ -177,12 +178,13 @@ func (TrinoPoolInstance) TableName() string { return "duckgres_trino_pool_instan
 // View projects the row onto the planner's read-only input.
 func (i TrinoPoolInstance) View() trinopool.InstanceView {
 	return trinopool.InstanceView{
-		ID:        i.InstanceID,
-		Phase:     trinopool.Phase(i.Phase),
-		ReleaseID: i.ReleaseID,
-		Repair:    i.Repair,
-		RepairFor: i.RepairFor,
-		CreatedAt: i.CreatedAt.UnixNano(),
+		ID:              i.InstanceID,
+		Phase:           trinopool.Phase(i.Phase),
+		ReleaseID:       i.ReleaseID,
+		Repair:          i.Repair,
+		RepairFor:       i.RepairFor,
+		NodeReplacement: i.NodeReplacementEvidence != nil,
+		CreatedAt:       i.CreatedAt.UnixNano(),
 	}
 }
 

controlplane/configstore/trino_pool_node_replacement.go

@@ -0,0 +1,33 @@
+package configstore
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/posthog/duckgres/controlplane/trinopool"
+	"gorm.io/gorm"
+)
+
+// RecordTrinoPoolNodeReplacement preserves the first infrastructure replacement request.
+// Retries and subsequent drift observations cannot rewrite the original node identity.
+func (cs *ConfigStore) RecordTrinoPoolNodeReplacement(ctx context.Context, lease TrinoPoolLease, instanceID string, evidence trinopool.NodeReplacementEvidence) error {
+	if err := evidence.Validate(); err != nil {
+		return err
+	}
+	encoded, err := json.Marshal(evidence)
+	if err != nil {
+		return err
+	}
+	return cs.withPoolAuthority(ctx, lease, func(tx *gorm.DB, _ *TrinoPool) error {
+		var instance TrinoPoolInstance
+		if err := tx.Where("instance_id = ? AND pool_id = ?", instanceID, lease.PoolID).First(&instance).Error; err != nil {
+			return fmt.Errorf("read instance for node replacement: %w", err)
+		}
+		if instance.NodeReplacementEvidence != nil || trinopool.Phase(instance.Phase).Terminal() {
+			return nil
+		}
+		return tx.Model(&TrinoPoolInstance{}).Where("instance_id = ? AND pool_id = ? AND node_replacement_evidence IS NULL", instanceID, lease.PoolID).
+			Update("node_replacement_evidence", string(encoded)).Error
+	})
+}

controlplane/trino_pool_config.go

@@ -26,9 +26,10 @@ import (
 // (`DUCKGRES_TRINO_POOL_CATALOG_CELL_ID`, see trino_pool_catalog.go) to match
 // what the cell's coordinators read.
 const (
-	envTrinoPoolEnabled         = "DUCKGRES_TRINO_POOL_ENABLED"
-	envTrinoPoolOperatorEnabled = "DUCKGRES_TRINO_POOL_OPERATOR_ENABLED"
-	envTrinoPoolGatewayURL      = "DUCKGRES_TRINO_POOL_GATEWAY_URL"
+	envTrinoPoolEnabled               = "DUCKGRES_TRINO_POOL_ENABLED"
+	envTrinoPoolOperatorEnabled       = "DUCKGRES_TRINO_POOL_OPERATOR_ENABLED"
+	envTrinoPoolGatewayURL            = "DUCKGRES_TRINO_POOL_GATEWAY_URL"
+	envTrinoPoolNodeDisruptionEnabled = "DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED"
 
 	trinoPoolModeFixed  = "fixed"
 	trinoPoolModeShared = "shared-pool"

controlplane/trino_pool_node_disruption.go

@@ -0,0 +1,143 @@
+//go:build kubernetes
+
+package controlplane
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/posthog/duckgres/controlplane/configstore"
+	"github.com/posthog/duckgres/controlplane/trinogateway"
+	"github.com/posthog/duckgres/controlplane/trinopool"
+)
+
+type trinoPoolNodeReplacementStore interface {
+	RecordTrinoPoolNodeReplacement(context.Context, configstore.TrinoPoolLease, string, trinopool.NodeReplacementEvidence) error
+}
+
+var errTrinoCandidateNodeReplacement = errors.New("candidate node requires replacement before admission")
+var errTrinoCandidateNodeRetired = errors.New("candidate retired before admission because its node requires replacement")
+
+func definitiveTrinoNodePlacement(err error) bool {
+	return errors.Is(err, errTrinoCandidateNodeReplacement) || errors.Is(err, errTrinoUnschedulableNode) || errors.Is(err, errTrinoNodePlacementInvalid)
+}
+
+// protectPoolNodes visits all live instances before any instance advances.
+// A busy candidate cannot starve the protection of existing serving pods.
+func (o *trinoPoolOperator) protectPoolNodes(ctx context.Context, instances []configstore.TrinoPoolInstance) error {
+	o.nodeProtectionErrors = make(map[string]error)
+	if o.nodeGuard == nil {
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	o.nodeGuard.reset()
+	o.nodeGuard.epoch = o.lease.Epoch
+	store, ok := o.store.(trinoPoolNodeReplacementStore)
+	if !ok {
+		return errors.New("node disruption handling requires durable replacement storage")
+	}
+	var failures []error
+	start := int(o.nodeProtectionCursor % uint64(max(1, len(instances))))
+	o.nodeProtectionCursor++
+	for offset := range instances {
+		index := (start + offset) % len(instances)
+		instance := &instances[index]
+		switch trinopool.Phase(instance.Phase) {
+		case trinopool.PhasePending, trinopool.PhaseRetiring, trinopool.PhaseRetired, trinopool.PhaseFailureRetired, trinopool.PhaseFailedPreparing, trinopool.PhaseLost:
+			continue
+		}
+		evidence, err := o.nodeGuard.inspect(ctx, inventoryOf(*instance))
+		if err == errTrinoUnschedulableNode && instance.NodeReplacementEvidence != nil {
+			err = nil
+		}
+		if len(evidence) > 0 {
+			// Exact positive evidence remains valid when another pod or node is unavailable.
+			recordErr := o.dropAuthority(store.RecordTrinoPoolNodeReplacement(ctx, o.lease, instance.InstanceID, evidence[0]))
+			if recordErr == nil && instance.NodeReplacementEvidence == nil {
+				encoded, _ := json.Marshal(evidence[0])
+				value := string(encoded)
+				instance.NodeReplacementEvidence = &value
+				slog.Info("Trino pool requested voluntary node replacement.", "pool", o.config.PublicID, "instance", instance.InstanceID, "node_uid", evidence[0].NodeUID, "nodeclaim_uid", evidence[0].NodeClaimUID, "reason", evidence[0].Reason)
+			}
+			if recordErr != nil {
+				err = errors.Join(err, recordErr)
+			}
+			if err == nil {
+				for _, node := range evidence {
+					if err = o.nodeGuard.cordon(ctx, node, o.config.PoolID, instance.InstanceID); err != nil {
+						break
+					}
+				}
+			}
+		}
+		if err == errTrinoPodProtectionIncomplete && (instance.Phase == string(trinopool.PhaseCreating) || instance.Phase == string(trinopool.PhasePreparing)) {
+			// Pending pods are normal convergence, not failed protection of serving work.
+			continue
+		}
+		if err != nil {
+			o.nodeProtectionErrors[instance.InstanceID] = err
+			failures = append(failures, fmt.Errorf("protect instance %s from node disruption: %w", instance.InstanceID, err))
+			if o.fenced {
+				return errors.Join(failures...)
+			}
+		}
+	}
+	return errors.Join(failures...)
+}
+
+func (o *trinoPoolOperator) candidateNodeCheck(ctx context.Context, instance configstore.TrinoPoolInstance) error {
+	if o.nodeGuard == nil {
+		return nil
+	}
+	if instance.NodeReplacementEvidence != nil {
+		return errTrinoCandidateNodeReplacement
+	}
+	o.nodeGuard.reset()
+	defer o.nodeGuard.reset()
+	o.nodeGuard.namedClaims = true
+	o.nodeGuard.epoch = o.lease.Epoch
+	evidence, err := o.nodeGuard.inspect(ctx, inventoryOf(instance))
+	if err != nil {
+		return err
+	}
+	if len(evidence) != 0 || instance.NodeReplacementEvidence != nil {
+		return errTrinoCandidateNodeReplacement
+	}
+	return nil
+}
+
+// nodeSafeAdmission resolves committed admission before inspecting new infrastructure facts.
+// A failed fresh check cannot erase an admission whose response was lost.
+func (o *trinoPoolOperator) nodeSafeAdmission(ctx context.Context, instance configstore.TrinoPoolInstance, request trinogateway.AdmitMemberRequest) (trinogateway.Member, error) {
+	if o.nodeGuard != nil {
+		current, err := o.gateway.GetMember(ctx, o.config.RoutingGroup, instance.InstanceID)
+		if err != nil {
+			return trinogateway.Member{}, err
+		}
+		if current.Incarnation != instance.GatewayIncarnation || current.PodUID != instance.CoordinatorPodUID || current.BootID != instance.CoordinatorBootID {
+			return trinogateway.Member{}, errors.New("admission read-back identity changed")
+		}
+		if current.Phase == "ACTIVE" {
+			return o.gateway.AdmitMember(ctx, o.config.RoutingGroup, instance.InstanceID, request)
+		}
+		if current.Phase == "RETIRING" || current.Phase == "RETIRED" {
+			return trinogateway.Member{}, errTrinoCandidateNodeRetired
+		}
+		if err := o.candidateNodeCheck(ctx, instance); err != nil {
+			if definitiveTrinoNodePlacement(err) {
+				_, retireErr := o.gateway.RetireMember(ctx, o.config.RoutingGroup, instance.InstanceID, trinogateway.MemberStepRequest{Step: o.step(instance.InstanceID, "retire"), ExpectedGeneration: current.Generation})
+				if retireErr != nil {
+					return trinogateway.Member{}, retireErr
+				}
+				return trinogateway.Member{}, errTrinoCandidateNodeRetired
+			}
+			return trinogateway.Member{}, err
+		}
+	}
+	return o.gateway.AdmitMember(ctx, o.config.RoutingGroup, instance.InstanceID, request)
+}