valkeycluster_controller.go

/*
Copyright 2025 Valkey Contributors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"reflect"
	"slices"
	"strconv"
	"strings"
	"time"

	valkeyiov1alpha1 "github.com/valkey-io/valkey-operator/api/v1alpha1"
	"github.com/valkey-io/valkey-operator/internal/valkey"
	corev1 "k8s.io/api/core/v1"
	policyv1 "k8s.io/api/policy/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/tools/events"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
)

const (
	DefaultPort           = 6379
	DefaultClusterBusPort = 16379
	DefaultImage          = "valkey/valkey:9.0.0"
	DefaultExporterImage  = "oliver006/redis_exporter:v1.80.0"
	DefaultExporterPort   = 9121

	// AclSecretType is the Secret type used for operator-managed ACL Secrets.
	AclSecretType = corev1.SecretType("valkey.io/acl")

	// Error messages
	statusUpdateFailedMsg = "failed to update status"
)

// ValkeyClusterReconciler reconciles a ValkeyCluster object
type ValkeyClusterReconciler struct {
	client.Client
	APIReader client.Reader
	Scheme    *runtime.Scheme
	Recorder  events.EventRecorder
}

// +kubebuilder:rbac:groups=valkey.io,resources=valkeyclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=valkey.io,resources=valkeyclusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=valkey.io,resources=valkeyclusters/finalizers,verbs=update
// +kubebuilder:rbac:groups=valkey.io,resources=valkeynodes,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch
// +kubebuilder:rbac:groups="apps",resources=deployments,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=events.k8s.io,resources=events,verbs=create;patch
// +kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=get;list;watch;create;update;patch;delete

// Reconcile is the main reconciliation loop. On each invocation it drives the
// cluster one step closer to the desired state described by the ValkeyCluster
// spec. The pipeline runs in the following order:
//
//   - Ensure the headless Service exists (upsertService).
//   - Ensure PodDisruptionBudget exists (reconcilePodDisruptionBudget).
//   - Ensure internal ACL users are configured (reconcileUsersAcl).
//   - Ensure the ConfigMap with valkey.conf and health-check scripts exists
//     (upsertConfigMap).
//   - Ensure one ValkeyNode per (shard, node) pair exists, creating missing
//     nodes and propagating spec changes one at a time in shard order with
//     replicas updated before the primary (reconcileValkeyNodes).
//   - Build the Valkey cluster state by connecting to each node and scraping
//     CLUSTER INFO / CLUSTER NODES.
//   - Promote orphaned replicas via CLUSTER FAILOVER TAKEOVER when quorum
//     is lost (promoteOrphanedReplicas).
//   - Forget stale nodes that no longer have a backing ValkeyNode.
//   - MEET: batch-introduce all isolated pending nodes to the cluster via
//     CLUSTER MEET. Requeue to let gossip propagate.
//   - Assign slots: batch-assign hash-slot ranges to all primary-labeled
//     pending nodes via CLUSTER ADDSLOTSRANGE.
//   - Replicate: batch-attach all replica-labeled pending nodes to their
//     matching primaries via CLUSTER REPLICATE.
//   - Scale-in: if the cluster has more shards than desired, drain slots
//     from excess shards via CLUSTER MIGRATESLOTS and delete their
//     ValkeyNodes once fully drained.
//   - Verify that the expected number of shards and replicas exist.
//   - Verify that all 16384 hash slots are assigned.
//   - If everything is healthy, mark the cluster Ready and requeue after 30s
//     for periodic health checks.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.23.3/pkg/reconcile
func (r *ValkeyClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { //nolint:gocyclo
	log := logf.FromContext(ctx)
	log.V(1).Info("reconcile...")

	cluster := &valkeyiov1alpha1.ValkeyCluster{}
	if err := r.Get(ctx, req.NamespacedName, cluster); err != nil {
		if apierrors.IsNotFound(err) {
			deleteClusterMetrics(req.Name, req.Namespace)
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, err
	}

	initClusterMetrics(req.Name, req.Namespace)

	if err := r.upsertService(ctx, cluster); err != nil {
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonServiceError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, err
	}

	if err := r.reconcilePodDisruptionBudget(ctx, cluster); err != nil {
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonPodDisruptionBudgetError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, err
	}

	if err := r.reconcileUsersAcl(ctx, cluster); err != nil {
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonUsersAclError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, err
	}

	if err := r.upsertConfigMap(ctx, cluster); err != nil {
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonConfigMapError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, err
	}
	// Roll hash ignores live-settable keys, so a change confined to those keys
	// does not roll the pods (they are applied live by the ValkeyNode controller).
	// Computed directly from cluster.Spec rather than reading back from the
	// ConfigMap to avoid a race condition where the cache does not have the ConfigMap
	configHash := serverConfigRollHash(cluster)

	// Surface a ConfigurationWarning condition when an explicit
	// terminationGracePeriodSeconds is too short for the graceful failover on
	// SIGTERM to finish before SIGKILL. The value is honoured; the operator does
	// not silently override it. The condition is idempotent, so the event fires
	// only when the cluster first enters the warning state.
	rec := recommendedGracePeriodSeconds(cluster)
	if g := cluster.Spec.TerminationGracePeriodSeconds; g != nil && *g < rec {
		msg := fmt.Sprintf("spec.terminationGracePeriodSeconds (%ds) is below the recommended %ds for cluster-manual-failover-timeout; SIGKILL may interrupt the graceful failover on shutdown", *g, rec)
		if !meta.IsStatusConditionTrue(cluster.Status.Conditions, valkeyiov1alpha1.ConditionConfigurationWarning) {
			log.Info("terminationGracePeriodSeconds is below the recommended minimum for graceful failover",
				"requested", *g, "recommended", rec)
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, valkeyiov1alpha1.ReasonGracePeriodTooShort, "ReconcileValkeyCluster", "%s", msg)
		}
		setCondition(cluster, valkeyiov1alpha1.ConditionConfigurationWarning, valkeyiov1alpha1.ReasonGracePeriodTooShort, msg, metav1.ConditionTrue)
	} else {
		removeConditionIfReason(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionConfigurationWarning, valkeyiov1alpha1.ReasonGracePeriodTooShort)
	}

	nodes := &valkeyiov1alpha1.ValkeyNodeList{}
	if err := r.List(ctx, nodes, client.InNamespace(cluster.Namespace), client.MatchingLabels(map[string]string{LabelCluster: cluster.Name})); err != nil {
		log.Error(err, "failed to list ValkeyNodes")
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonValkeyNodeListError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, err
	}

	if requeue, err := r.reconcileValkeyNodes(ctx, cluster, nodes, configHash); err != nil {
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonValkeyNodeError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, err
	} else if requeue {
		if result, handled, err := r.handlePodSchedulingIssues(ctx, cluster); err != nil {
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonValkeyNodeError, err.Error(), metav1.ConditionFalse)
			_ = r.updateStatus(ctx, cluster, nil)
			return ctrl.Result{}, err
		} else if handled {
			return result, nil
		}
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonUpdatingNodes, "Updating ValkeyNodes", metav1.ConditionFalse)
		setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonUpdatingNodes, "Updating ValkeyNodes", metav1.ConditionTrue)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
	}
	operatorPassword, err := fetchSystemUserPassword(ctx, operatorUser, r.Client, cluster.Name, cluster.Namespace)
	if err != nil {
		log.Error(err, "failed to retrieve system user password")
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonSystemUsersAclError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, nil)
		return ctrl.Result{}, nil
	}
	state := r.getValkeyClusterState(ctx, cluster, nodes, operatorUser, operatorPassword)
	defer state.CloseClients()

	// Promote replicas of dead primaries when quorum is lost.
	// TAKEOVER before FORGET so slots remain continuously owned.
	if result, handled := r.promoteOrphanedReplicas(ctx, cluster, state); handled {
		return result, nil
	}

	r.forgetStaleNodes(ctx, cluster, state, nodes)

	// --- Phase 1: MEET all isolated nodes in one batch ---
	// A node with cluster_known_nodes <= 1 hasn't been introduced to the
	// cluster yet. CLUSTER MEET is idempotent and has no ordering
	// dependencies, so we issue it for every isolated pending node in a
	// single reconcile pass. Phase 2 refuses to assign slots to isolated
	// nodes, so every node is guaranteed to pass through here first.
	// After MEET, we requeue to let gossip propagate before proceeding
	// to slot assignment or replication.
	{
		met, err := r.meetIsolatedNodes(ctx, cluster, state)
		if err != nil {
			setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonNodeAddFailed, err.Error(), metav1.ConditionTrue)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
		if met > 0 {
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ClusterMeetBatch", "ClusterMeet", "Introduced %d isolated node(s) to the cluster", met)
			setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonAddingNodes, "Introducing nodes to cluster", metav1.ConditionTrue)
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Cluster is Reconciling", metav1.ConditionFalse)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
	}

	// --- Phase 2: Assign slots to all primary-labeled pending nodes ---
	// Slot assignment (CLUSTER ADDSLOTSRANGE) makes a pending node a
	// slot-owning primary. During scale-out (no unassigned slots), this
	// is a no-op; new primaries stay in PendingNodes until the rebalancer
	// migrates slots to them.
	if len(state.PendingNodes) > 0 {
		assigned, err := r.assignSlotsToPendingPrimaries(ctx, cluster, state, nodes)
		if err != nil {
			setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonNodeAddFailed, err.Error(), metav1.ConditionTrue)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
		if assigned > 0 {
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "PrimariesCreated", "AssignSlots", "Assigned slots to %d new primary node(s)", assigned)
			setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonAddingNodes, "Assigning slots to primaries", metav1.ConditionTrue)
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Cluster is Reconciling", metav1.ConditionFalse)
			_ = r.updateStatus(ctx, cluster, state)
			// Requeue so that the next reconcile sees the newly-created shards
			// in state.Shards, which replicas need for their lookup.
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
	}

	// --- Phase 3: REPLICATE all replica-labeled pending nodes ---
	// By this point all currently known primaries have slots and appear in state.Shards.
	// CLUSTER REPLICATE for different replicas targets different primaries,
	// so they can all be issued in one pass.
	if len(state.PendingNodes) > 0 {
		replicated, err := r.replicatePendingReplicas(ctx, cluster, state, nodes)
		if err != nil {
			setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonNodeAddFailed, err.Error(), metav1.ConditionTrue)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
		if replicated > 0 {
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ReplicasAttached", "CreateReplica", "Attached %d replica node(s)", replicated)
			setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonAddingNodes, "Attaching replicas", metav1.ConditionTrue)
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Cluster is Reconciling", metav1.ConditionFalse)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
	}

	// Build the effective shard list: state.Shards plus any pending primaries
	// that are scale-out leaders. During scale-out, GetClusterState places
	// new slot-less primaries in PendingNodes because it can't distinguish
	// them from unreplicated replicas. We use pod labels to identify them
	// and include them as empty shards for health checks and rebalancing.
	allShards := effectiveShards(state, nodes)

	// Handle scale-in: drain excess shards and clean up leftover ValkeyNodes.
	// This runs before pod scheduling checks so that excess shard pods from a
	// prior scale-up don't block scale-down when they can't be scheduled.
	if result, requeue := r.handleScaleIn(ctx, cluster, state, nodes); requeue {
		return result, nil
	}

	if result, handled, err := r.handlePodSchedulingIssues(ctx, cluster); err != nil {
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonValkeyNodeError, err.Error(), metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, state)
		return ctrl.Result{}, err
	} else if handled {
		return result, nil
	}

	// Check cluster status
	if len(allShards) < int(cluster.Spec.Shards) {
		log.V(1).Info("missing shards, requeue..")
		r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "WaitingForShards", "CheckShards", "%d of %d shards exist", len(allShards), cluster.Spec.Shards)
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonMissingShards, "Waiting for all shards to be created", metav1.ConditionFalse)
		setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconciling, "Creating shards", metav1.ConditionTrue)
		setCondition(cluster, valkeyiov1alpha1.ConditionClusterFormed, valkeyiov1alpha1.ReasonMissingShards, "Waiting for shards", metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, state)
		return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
	}
	for _, shard := range allShards {
		if countSlots(shard.Slots) == 0 {
			continue
		}
		if len(shard.Nodes) < (1 + int(cluster.Spec.Replicas)) {
			log.V(1).Info("missing replicas, requeue..")
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "WaitingForReplicas", "CheckReplicas", "Shard has %d of %d nodes", len(shard.Nodes), 1+int(cluster.Spec.Replicas))
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonMissingReplicas, "Waiting for all replicas to be created", metav1.ConditionFalse)
			setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconciling, "Creating replicas", metav1.ConditionTrue)
			setCondition(cluster, valkeyiov1alpha1.ConditionClusterFormed, valkeyiov1alpha1.ReasonMissingReplicas, "Waiting for replicas", metav1.ConditionFalse)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
	}

	// Check if all slots are assigned
	unassignedSlots := state.GetUnassignedSlots()
	allSlotsAssigned := len(unassignedSlots) == 0
	if !allSlotsAssigned {
		log.V(1).Info("slots are not assigned, requeue..", "unassignedSlots", unassignedSlots)
		setCondition(cluster, valkeyiov1alpha1.ConditionSlotsAssigned, valkeyiov1alpha1.ReasonSlotsUnassigned, "Waiting for slots to be assigned", metav1.ConditionFalse)
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Waiting for all slots to be assigned", metav1.ConditionFalse)
		setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconciling, "Waiting for slots to be assigned", metav1.ConditionTrue)
		setCondition(cluster, valkeyiov1alpha1.ConditionClusterFormed, valkeyiov1alpha1.ReasonSlotsUnassigned, "Waiting for slots to be assigned", metav1.ConditionFalse)
		_ = r.updateStatus(ctx, cluster, state)
		return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
	}

	// Check that all replicas have their replication link up (master_link_status:up).
	// before marking the cluster Ready, we need to make sure all replicas are in sync with their primary.
	for _, shard := range allShards {
		for _, node := range shard.Nodes {
			if !node.IsReplicationInSync() {
				log.V(1).Info("replica not yet in sync, requeue..", "address", node.Address)
				setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Waiting for replicas to sync with primary", metav1.ConditionFalse)
				setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconciling, "Waiting for replica sync", metav1.ConditionTrue)
				_ = r.updateStatus(ctx, cluster, state)
				return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
			}
		}
	}

	// --- Rebalance slots across primaries (scale-out) ---
	// After all shards are healthy and all slots are assigned, check if
	// the slot distribution is uneven (e.g. a new shard was added with
	// zero slots). rebalanceSlots picks a single src→dst move per
	// reconcile and migrates up to rebalanceSlotBatchSize slots at a
	// time using CLUSTER MIGRATESLOTS (atomic migration, requires
	// Valkey >= 9.0). Each pass requeues so the next reconcile
	// re-evaluates cluster state with fresh topology before continuing.
	rebalanced, err := r.rebalanceSlots(ctx, cluster, allShards)
	if err != nil {
		log.Error(err, "slot rebalancing failed")
		r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "SlotRebalanceFailed", "RebalanceSlots", "Slot rebalancing failed: %v", err)
		setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonRebalanceFailed, err.Error(), metav1.ConditionTrue)
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Cluster is Reconciling", metav1.ConditionFalse)
		setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonRebalancingSlots, "Rebalancing slots across primaries", metav1.ConditionTrue)
		_ = r.updateStatus(ctx, cluster, state)
		return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
	}
	if rebalanced {
		// Clear any stale Degraded condition from earlier phases (e.g.
		// NodeAddFailed set when replicas couldn't attach to primaries
		// that hadn't received slots yet during scale-out).
		meta.RemoveStatusCondition(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionDegraded)
		setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Cluster is Reconciling", metav1.ConditionFalse)
		setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonRebalancingSlots, "Rebalancing slots across primaries", metav1.ConditionTrue)
		_ = r.updateStatus(ctx, cluster, state)
		return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
	}

	// Cluster is healthy - set all positive conditions
	r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ClusterReady", "ReconcileCluster", "Cluster ready with %d shards and %d replicas", cluster.Spec.Shards, cluster.Spec.Replicas)
	setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonClusterHealthy, "Cluster is healthy", metav1.ConditionTrue)
	setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconcileComplete, "No changes needed", metav1.ConditionFalse)
	meta.RemoveStatusCondition(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionDegraded)
	setCondition(cluster, valkeyiov1alpha1.ConditionClusterFormed, valkeyiov1alpha1.ReasonTopologyComplete, "All nodes joined cluster", metav1.ConditionTrue)
	setCondition(cluster, valkeyiov1alpha1.ConditionSlotsAssigned, valkeyiov1alpha1.ReasonAllSlotsAssigned, "All slots assigned", metav1.ConditionTrue)

	if err := r.updateStatus(ctx, cluster, state); err != nil {
		log.Error(err, statusUpdateFailedMsg)
		return ctrl.Result{}, err
	}

	log.V(1).Info("reconcile done")
	return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
}

type podSchedulingIssue struct {
	podName string
	message string
}

func (r *ValkeyClusterReconciler) handlePodSchedulingIssues(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster) (ctrl.Result, bool, error) {
	issue, err := r.findPodSchedulingIssue(ctx, cluster)
	if err != nil {
		return ctrl.Result{}, false, err
	}
	if issue == nil {
		removeConditionIfReason(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonPodUnschedulable)
		removeConditionIfReason(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonPodUnschedulable)
		return ctrl.Result{}, false, nil
	}

	message := fmt.Sprintf("Pod %s is unschedulable", issue.podName)
	if issue.message != "" {
		message = fmt.Sprintf("%s: %s", message, issue.message)
	}
	r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "PodUnschedulable", "SchedulePod", "%s", message)
	setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonPodUnschedulable, message, metav1.ConditionTrue)
	setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonPodUnschedulable, message, metav1.ConditionFalse)
	setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconciling, "Waiting for unschedulable pods to be scheduled", metav1.ConditionTrue)
	if err := r.updateStatus(ctx, cluster, nil); err != nil {
		return ctrl.Result{}, false, err
	}
	return ctrl.Result{RequeueAfter: 10 * time.Second}, true, nil
}

func (r *ValkeyClusterReconciler) findPodSchedulingIssue(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster) (*podSchedulingIssue, error) {
	pods := &corev1.PodList{}
	if err := r.List(ctx, pods, client.InNamespace(cluster.Namespace), client.MatchingLabels(map[string]string{LabelCluster: cluster.Name})); err != nil {
		return nil, fmt.Errorf("list Valkey pods: %w", err)
	}

	desiredShards := int(cluster.Spec.Shards)
	for i := range pods.Items {
		if label, ok := pods.Items[i].Labels[LabelShardIndex]; ok {
			si, err := strconv.Atoi(label)
			if err == nil && si >= desiredShards {
				continue
			}
		}
		if issue := podSchedulingIssueForPod(&pods.Items[i]); issue != nil {
			return issue, nil
		}
	}
	return nil, nil
}

func podSchedulingIssueForPod(pod *corev1.Pod) *podSchedulingIssue {
	if pod.DeletionTimestamp != nil {
		return nil
	}
	for _, condition := range pod.Status.Conditions {
		if condition.Type == corev1.PodScheduled &&
			condition.Status == corev1.ConditionFalse &&
			condition.Reason == corev1.PodReasonUnschedulable {
			return &podSchedulingIssue{
				podName: pod.Name,
				message: condition.Message,
			}
		}
	}
	return nil
}

func headlessServiceName(clusterName string) string {
	return resourcePrefix + clusterName
}

// Create or update a headless service (client connects to pods directly)
func (r *ValkeyClusterReconciler) upsertService(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster) error {
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      headlessServiceName(cluster.Name),
			Namespace: cluster.Namespace,
		},
	}
	result, err := controllerutil.CreateOrUpdate(ctx, r.Client, svc, func() error {
		svc.Labels = labels(cluster)
		svc.Spec.Type = corev1.ServiceTypeClusterIP
		// ClusterIP is immutable after creation; preserve the existing value on updates.
		if svc.Spec.ClusterIP == "" {
			svc.Spec.ClusterIP = "None"
		}
		svc.Spec.Selector = map[string]string{LabelCluster: cluster.Name}
		// Protocol and TargetPort are API-server defaults; set them explicitly
		// so the rebuilt slice deep-equals the stored one and CreateOrUpdate
		// stops updating the Service on every reconcile (#315).
		svc.Spec.Ports = []corev1.ServicePort{{
			Name:       appName,
			Port:       DefaultPort,
			Protocol:   corev1.ProtocolTCP,
			TargetPort: intstr.FromInt32(DefaultPort),
		}}
		return controllerutil.SetControllerReference(cluster, svc, r.Scheme)
	})
	if err != nil {
		r.Recorder.Eventf(cluster, svc, corev1.EventTypeWarning, "ServiceUpdateFailed", "UpdateService", "Failed to upsert Service: %v", err)
		return err
	}
	if result == controllerutil.OperationResultCreated {
		r.Recorder.Eventf(cluster, svc, corev1.EventTypeNormal, "ServiceCreated", "CreateService", "Created headless Service")
	}
	return nil
}

// reconcileValkeyNodes ensures every (shard, nodeIndex) pair has a ValkeyNode CR.
// Each ValkeyNode manages exactly one Pod (Replicas=1) and is named
// deterministically:
//
//	<cluster>-<N>-<M>
//
// where N is the shard index and M is the node index (0 = initial primary,
// 1+ = replicas). It iterates shards in ascending order and nodes replica-first within each
// shard, using live cluster state to identify the actual primary. At most one spec update is issued per reconcile.
// Once a node is updated or found not-ready after a prior update,
// the function returns (true, nil) so the caller requeues before advancing to the next node.
//
// For a 3-shard cluster with 2 replicas per shard, this produces 9 ValkeyNodes:
//
//	mycluster-0-0, mycluster-0-1, mycluster-0-2,
//	mycluster-1-0, mycluster-1-1, mycluster-1-2,
//	mycluster-2-0, mycluster-2-1, mycluster-2-2.
func (r *ValkeyClusterReconciler) reconcileValkeyNodes(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, nodes *valkeyiov1alpha1.ValkeyNodeList, configHash string) (bool, error) {
	log := logf.FromContext(ctx)

	nodesPerShard := 1 + int(cluster.Spec.Replicas)
	totalCreated := 0

	// Scrape cluster state once for proactive failover decisions, but only
	// when at least one node actually needs a roll. During initial bootstrap
	// no nodes exist, so state stays nil. The snapshot is safe to reuse
	// across the loop: replicaFirstNodeOrder uses it to place the actual
	// primary last within each shard, and after an update we requeue
	// immediately, re-scraping fresh state before any further rolls.
	var clusterState *valkey.ClusterState
	// Scrape topology for Spec rolls and for workload-template rolls (operator
	// builder drift) so proactive failover still runs before killing a primary.
	if anyNodeRequiresRoll(cluster, nodes, configHash) || anyNodeRequiresWorkloadRoll(nodes) {
		operatorPassword, err := fetchSystemUserPassword(ctx, operatorUser, r.Client, cluster.Name, cluster.Namespace)
		if err != nil {
			return false, fmt.Errorf("failed to fetch operator password for proactive failover: %w", err)
		}
		clusterState = r.getValkeyClusterState(ctx, cluster, nodes, operatorUser, operatorPassword)
		defer clusterState.CloseClients()
	}

	for shardIndex := range int(cluster.Spec.Shards) {
		// If rolls are in progress but the primary of an active shard cannot be
		// identified from cluster state, defer rather than roll in an unknown order.
		// New shards (not yet in the topology) are exempt — they need creation, not rolling.
		if clusterState != nil && shardExistsInTopology(clusterState, shardIndex, nodes) &&
			primaryNodeIndexForShard(shardIndex, nodesPerShard, nodes, clusterState) < 0 {
			log.Info("cannot identify primary for shard, deferring roll", "shardIndex", shardIndex)
			return true, nil
		}
		// Iterate nodes replica-first: use live cluster state to identify the
		// actual primary (which may differ from node-index=0 after a failover)
		// and place it last.
		for _, nodeIndex := range replicaFirstNodeOrder(shardIndex, nodesPerShard, nodes, clusterState) {
			result, err := r.reconcileValkeyNode(ctx, cluster, shardIndex, nodeIndex, clusterState, configHash)
			if err != nil {
				return false, err
			}
			switch result {
			case nodeDeferred:
				// Roll deferred; MEET and REPLICATE isolated nodes so they
				// rejoin the shard before next reconcile.
				if _, meetErr := r.meetIsolatedNodes(ctx, cluster, clusterState); meetErr != nil {
					log.Error(meetErr, "meetIsolatedNodes failed during deferred roll")
				}
				if _, replErr := r.replicatePendingReplicas(ctx, cluster, clusterState, nodes); replErr != nil {
					log.Error(replErr, "replicatePendingReplicas failed during deferred roll")
				}
				return true, nil
			case nodeRequeued:
				return true, nil
			case nodeCreated:
				totalCreated++
			}
		}
	}

	if totalCreated > 0 {
		log.V(1).Info("created ValkeyNodes", "count", totalCreated)
	}
	return false, nil
}

// nodeResult describes the outcome of reconciling a single ValkeyNode.
type nodeResult int

const (
	nodeUnchanged nodeResult = iota // no spec change, node is ready
	nodeRequeued                    // spec updated or not ready, caller should requeue
	nodeCreated                     // new ValkeyNode was created
	nodeDeferred                    // primary roll deferred, waiting for synced replica
)

// reconcileValkeyNode reconciles a single ValkeyNode for (shardIndex, nodeIndex).
// Returns a nodeResult signaling the outcome or required next action.
func (r *ValkeyClusterReconciler) reconcileValkeyNode(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, shardIndex, nodeIndex int, clusterState *valkey.ClusterState, configHash string) (nodeResult, error) {
	log := logf.FromContext(ctx)

	desired := buildClusterValkeyNode(cluster, shardIndex, nodeIndex)
	desired.Spec.ServerConfigHash = configHash
	node := &valkeyiov1alpha1.ValkeyNode{
		ObjectMeta: metav1.ObjectMeta{
			Name:      desired.Name,
			Namespace: desired.Namespace,
		},
	}

	// Load current node (if any) for failover + workload-drift decisions.
	current := &valkeyiov1alpha1.ValkeyNode{}
	currentExists := true
	if err := r.Get(ctx, client.ObjectKeyFromObject(node), current); err != nil {
		if !apierrors.IsNotFound(err) {
			return nodeUnchanged, err
		}
		currentExists = false
	}

	needsSpecRoll := currentExists && nodeRequiresRoll(current, desired)
	needsWorkloadPermit := currentExists && nodeNeedsWorkloadPermit(current)
	inFlightWorkloadRoll := currentExists && nodeHasInFlightWorkloadRoll(current)
	// Failover / ordered roll applies to Spec changes and workload template rolls.
	needsWorkloadRoll := needsWorkloadPermit || inFlightWorkloadRoll

	if deferred := r.maybeProactiveFailoverBeforeRoll(ctx, cluster, clusterState, current, needsSpecRoll || needsWorkloadRoll); deferred {
		return nodeDeferred, nil
	}

	// One-at-a-time for Spec and workload rolls: never grant a new permit or
	// Spec update while another node still holds an in-flight allow annotation.
	if needsSpecRoll || needsWorkloadPermit {
		if other, err := r.otherNodeHasInFlightWorkloadRoll(ctx, cluster, node.Name); err != nil {
			return nodeUnchanged, err
		} else if other {
			log.V(1).Info("another ValkeyNode already holds a workload roll permit, waiting", "name", node.Name)
			return nodeRequeued, nil
		}
	}

	result, err := controllerutil.CreateOrUpdate(ctx, r.Client, node, func() error {
		node.Labels = desired.Labels
		node.Spec = desired.Spec
		// Spec-driven rolls: grant a one-shot permit so the node controller may
		// apply the new pod template. Workload-only drift uses the desired hash
		// path below (CreateOrUpdate result None).
		if needsSpecRoll {
			setNodeAnnotation(node, allowWorkloadRevisionAnnotation, allowWorkloadRevisionAny)
		}
		return controllerutil.SetControllerReference(cluster, node, r.Scheme)
	})
	if err != nil {
		r.Recorder.Eventf(cluster, node, corev1.EventTypeWarning, "ValkeyNodeFailed", "ReconcileValkeyNode", "Failed to reconcile ValkeyNode: %v", err)
		return nodeUnchanged, err
	}
	switch result {
	case controllerutil.OperationResultCreated:
		r.Recorder.Eventf(cluster, node, corev1.EventTypeNormal, "ValkeyNodeCreated", "CreateValkeyNode", "Created ValkeyNode for shard %d node %d", shardIndex, nodeIndex)
		return nodeCreated, nil
	case controllerutil.OperationResultUpdated:
		// A spec change was applied. Requeue unconditionally so the node has
		// time to settle before we advance to the next one (one-at-a-time
		// rolling update).
		log.V(1).Info("updated ValkeyNode, waiting for it to become ready", "name", node.Name)
		r.Recorder.Eventf(cluster, node, corev1.EventTypeNormal, "ValkeyNodeUpdated", "UpdateValkeyNode", "Updated ValkeyNode %s", node.Name)
		return nodeRequeued, nil
	case controllerutil.OperationResultNone:
		return r.handleUnchangedValkeyNode(ctx, cluster, node, inFlightWorkloadRoll, needsWorkloadPermit)
	default:
		log.V(1).Info("unexpected CreateOrUpdate result", "result", result, "name", node.Name)
	}
	return nodeUnchanged, nil
}

// maybeProactiveFailoverBeforeRoll runs proactive failover when rolling a
// primary. Returns true when the primary has no synced replica yet (defer).
func (r *ValkeyClusterReconciler) maybeProactiveFailoverBeforeRoll(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, clusterState *valkey.ClusterState, current *valkeyiov1alpha1.ValkeyNode, rolling bool) bool {
	log := logf.FromContext(ctx)
	if clusterState == nil || !rolling {
		return false
	}
	shard, replicas := findFailoverShard(clusterState, current.Status.PodIP)
	if shard != nil {
		log.Info("proactive failover before rolling primary",
			"name", current.Name, "address", current.Status.PodIP,
			"syncedReplicas", len(replicas))
		if err := proactiveFailover(ctx, r.Recorder, cluster, shard, replicas); err != nil {
			log.Info("proactive failover did not complete, proceeding with roll",
				"name", current.Name, "err", err)
		}
		return false
	}
	if cluster.Spec.Replicas == 0 {
		return false
	}
	// findFailoverShard returned nil for one of three reasons:
	// 1. Node is the shard primary but has no synced replicas: wait for replica to rejoin
	// 2. Node is in a shard but is a replica: safe to roll
	// 3. Node isn't in any shard (isolated): safe to roll
	// Only case 1 requires waiting; identify it's the actual primary of its shard.
	shardInState := clusterState.FindShardForAddress(current.Status.PodIP)
	if shardInState != nil && shardInState.GetPrimaryNode() != nil && shardInState.GetPrimaryNode().Address == current.Status.PodIP {
		log.Info("primary has no synced replicas, deferring roll",
			"name", current.Name, "address", current.Status.PodIP,
			"shardNodes", len(shardInState.Nodes),
			"shardId", shardInState.Id)
		return true
	}
	return false
}

// handleUnchangedValkeyNode handles CreateOrUpdate OperationResultNone: grant
// a workload permit if needed, or wait until the node is settled and Ready.
func (r *ValkeyClusterReconciler) handleUnchangedValkeyNode(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, node *valkeyiov1alpha1.ValkeyNode, inFlightWorkloadRoll, needsWorkloadPermit bool) (nodeResult, error) {
	log := logf.FromContext(ctx)

	// In-flight permitted roll: wait until the node clears the permit
	// (after the workload revision is live) and is Ready.
	if inFlightWorkloadRoll {
		log.V(1).Info("ValkeyNode workload roll in flight, waiting", "name", node.Name)
		return nodeRequeued, nil
	}
	// Builder/template drift without Spec change: grant permit for this node.
	// Sibling in-flight was already checked by the caller.
	if needsWorkloadPermit {
		return r.grantWorkloadRollPermit(ctx, cluster, node)
	}
	if node.Status.ObservedGeneration > 0 && node.Generation != node.Status.ObservedGeneration {
		log.V(1).Info("ValkeyNode spec not yet observed by controller, waiting",
			"name", node.Name,
			"generation", node.Generation,
			"observedGeneration", node.Status.ObservedGeneration)
		return nodeRequeued, nil
	}
	if !node.Status.Ready {
		// No spec change, but the node hasn't reached Ready yet (e.g.
		// still starting after a prior update). Unlike Updated above, we
		// only wait when not-ready; a ready unchanged node is safe to
		// advance past.
		log.V(1).Info("ValkeyNode not yet ready, waiting", "name", node.Name)
		return nodeRequeued, nil
	}
	if c := meta.FindStatusCondition(node.Status.Conditions, valkeyiov1alpha1.ValkeyNodeConditionLiveConfigApplied); c != nil && c.Status == metav1.ConditionFalse {
		log.V(1).Info("ValkeyNode live config not yet applied, waiting", "name", node.Name)
		return nodeRequeued, nil
	}
	return nodeUnchanged, nil
}

// grantWorkloadRollPermit writes allow-workload-revision for a node waiting on
// builder/template drift. Always requeues so the node controller can apply.
func (r *ValkeyClusterReconciler) grantWorkloadRollPermit(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, node *valkeyiov1alpha1.ValkeyNode) (nodeResult, error) {
	log := logf.FromContext(ctx)
	desiredHash := ""
	if node.Annotations != nil {
		desiredHash = node.Annotations[desiredWorkloadRevisionAnnotation]
	}
	if desiredHash == "" {
		// Condition True but annotation not yet set; wait for node controller.
		log.V(1).Info("ValkeyNode workload drift pending without desired hash, waiting", "name", node.Name)
		return nodeRequeued, nil
	}
	if !workloadPermitAllows(node, desiredHash) {
		patchBase := node.DeepCopy()
		if setNodeAnnotation(node, allowWorkloadRevisionAnnotation, desiredHash) {
			if err := r.Patch(ctx, node, client.MergeFrom(patchBase)); err != nil {
				return nodeUnchanged, err
			}
			log.Info("granted workload roll permit", "name", node.Name, "hash", desiredHash)
			r.Recorder.Eventf(cluster, node, corev1.EventTypeNormal, "WorkloadRollPermitted", "PermitWorkloadRoll",
				"Granted workload roll permit for %s (hash %s)", node.Name, desiredHash)
		}
	}
	return nodeRequeued, nil
}

// otherNodeHasInFlightWorkloadRoll reports whether any ValkeyNode in this cluster
// other than excludeName currently holds a workload roll permit.
func (r *ValkeyClusterReconciler) otherNodeHasInFlightWorkloadRoll(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, excludeName string) (bool, error) {
	list := &valkeyiov1alpha1.ValkeyNodeList{}
	if err := r.List(ctx, list, client.InNamespace(cluster.Namespace), client.MatchingLabels{LabelCluster: cluster.Name}); err != nil {
		return false, err
	}
	for i := range list.Items {
		n := &list.Items[i]
		if n.Name == excludeName {
			continue
		}
		if nodeHasInFlightWorkloadRoll(n) {
			return true, nil
		}
	}
	return false, nil
}

const (
	// gracePeriodBufferSeconds is added on top of cluster-manual-failover-timeout
	// so the SIGTERM-triggered failover has headroom to finish before SIGKILL.
	gracePeriodBufferSeconds = 10
	// defaultFailoverTimeoutSeconds mirrors the Valkey default for
	// cluster-manual-failover-timeout (5000ms).
	defaultFailoverTimeoutSeconds = 5
	// defaultGracePeriodSeconds mirrors the Kubernetes default
	// terminationGracePeriodSeconds.
	defaultGracePeriodSeconds = 30
)

// failoverTimeoutSeconds returns cluster-manual-failover-timeout in whole
// seconds (rounded up), or the Valkey default when unset or unparseable.
func failoverTimeoutSeconds(config map[string]string) int64 {
	v, ok := config["cluster-manual-failover-timeout"]
	if !ok {
		return defaultFailoverTimeoutSeconds
	}
	ms, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64)
	if err != nil || ms <= 0 {
		return defaultFailoverTimeoutSeconds
	}
	return (ms + 999) / 1000
}

// recommendedGracePeriodSeconds is the smallest terminationGracePeriodSeconds
// that lets a SIGTERM-triggered failover finish before SIGKILL.
func recommendedGracePeriodSeconds(cluster *valkeyiov1alpha1.ValkeyCluster) int64 {
	return failoverTimeoutSeconds(cluster.Spec.Config) + gracePeriodBufferSeconds
}

// effectiveGracePeriodSeconds is the grace period the operator applies to the
// Valkey pods: the user's value when set, otherwise a safe default that is at
// least the Kubernetes default and the recommended minimum.
func effectiveGracePeriodSeconds(cluster *valkeyiov1alpha1.ValkeyCluster) int64 {
	if cluster.Spec.TerminationGracePeriodSeconds != nil {
		return *cluster.Spec.TerminationGracePeriodSeconds
	}
	if rec := recommendedGracePeriodSeconds(cluster); rec > defaultGracePeriodSeconds {
		return rec
	}
	return defaultGracePeriodSeconds
}

// buildClusterValkeyNode constructs the ValkeyNode CR for a given (shard, node) position.
func buildClusterValkeyNode(cluster *valkeyiov1alpha1.ValkeyCluster, shardIndex int, nodeIndex int) *valkeyiov1alpha1.ValkeyNode {
	// Start with recommended k8s labels; instance is the cluster name and component is "valkey-node".
	l := baseLabels(cluster.Name, "valkey-node")
	// Inherit user-defined labels from the parent cluster at lower priority.
	for k, v := range cluster.Labels {
		if _, exists := l[k]; !exists {
			l[k] = v
		}
	}
	// Operator-specific labels always take precedence.
	l[LabelCluster] = cluster.Name
	l[LabelShardIndex] = strconv.Itoa(shardIndex)
	l[LabelNodeIndex] = strconv.Itoa(nodeIndex)

	var scheduling valkeyiov1alpha1.SchedulingSpec
	if cluster.Spec.Scheduling != nil {
		scheduling = *cluster.Spec.Scheduling
	}

	// Only set the field when it differs from the Kubernetes default, so existing
	// clusters that resolve to the default keep a nil value and are not rolled on
	// operator upgrade.
	// Store the user's value verbatim when set, so an explicit change (including
	// back to the Kubernetes default) always propagates to the pod. When unset,
	// only populate the field if the derived default deviates from the Kubernetes
	// pod default, so existing clusters that resolve to the default are not rolled
	// on operator upgrade.
	var gracePeriod *int64
	if cluster.Spec.TerminationGracePeriodSeconds != nil {
		g := *cluster.Spec.TerminationGracePeriodSeconds
		gracePeriod = &g
	} else if d := effectiveGracePeriodSeconds(cluster); d != defaultGracePeriodSeconds {
		gracePeriod = &d
	}

	// Curate node-axis scheduling primitives from node.spread, merged with the
	// user's escape-hatch passthrough. Preserve nil when nothing is curated.
	shardMode, primariesMode, podsMode := effectiveNodeSpread(cluster.Spec.Scheduling)
	affinity := withNodeShardAntiAffinity(scheduling.Affinity, cluster.Name, shardIndex, shardMode)
	topologySpreadConstraints := scheduling.TopologySpreadConstraints
	if curated := nodeSpreadTSCs(cluster.Name, nodeIndex, primariesMode, podsMode); len(curated) > 0 {
		topologySpreadConstraints = append(
			append([]corev1.TopologySpreadConstraint{}, topologySpreadConstraints...),
			curated...,
		)
	}

	return &valkeyiov1alpha1.ValkeyNode{
		ObjectMeta: metav1.ObjectMeta{
			Name:      valkeyNodeName(cluster.Name, shardIndex, nodeIndex),
			Namespace: cluster.Namespace,
			Labels:    l,
		},
		Spec: valkeyiov1alpha1.ValkeyNodeSpec{
			Image:                         cluster.Spec.Image,
			ImagePullSecrets:              cluster.Spec.ImagePullSecrets,
			WorkloadType:                  cluster.Spec.WorkloadType,
			Persistence:                   cluster.Spec.Persistence,
			Resources:                     cluster.Spec.Resources,
			NodeSelector:                  scheduling.NodeSelector,
			Affinity:                      affinity,
			Tolerations:                   scheduling.Tolerations,
			PriorityClassName:             scheduling.PriorityClassName,
			TopologySpreadConstraints:     topologySpreadConstraints,
			Exporter:                      cluster.Spec.Exporter,
			Containers:                    cluster.Spec.Containers,
			ServerConfigMapName:           GetServerConfigMapName(cluster.Name),
			UsersACLSecretName:            getInternalSecretName(cluster.Name),
			TLS:                           cluster.Spec.TLS,
			Config:                        cluster.Spec.Config,
			PodSecurityContext:            cluster.Spec.PodSecurityContext,
			TerminationGracePeriodSeconds: gracePeriod,
		},
	}
}

func (r *ValkeyClusterReconciler) getValkeyClusterState(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, nodes *valkeyiov1alpha1.ValkeyNodeList, username, password string) *valkey.ClusterState {
	ips := []string{}
	for _, node := range nodes.Items {
		if node.Status.PodIP == "" {
			continue
		}
		ips = append(ips, node.Status.PodIP)
	}
	var tlsConfig *tls.Config
	if cluster.Spec.TLS != nil && cluster.Spec.TLS.Certificate.SecretName != "" {
		serverName := fmt.Sprintf("%s.%s.svc.cluster.local", headlessServiceName(cluster.Name), cluster.Namespace)
		cfg, err := getTLSConfig(ctx, r.APIReader, cluster.Spec.TLS.Certificate.SecretName, serverName, cluster.Namespace)
		if err == nil {
			tlsConfig = cfg
		}
	}
	return valkey.GetClusterState(ctx, ips, DefaultPort, username, password, tlsConfig)
}

// findMeetTarget picks the best node to MEET all isolated nodes against.
// Priority: (1) a shard primary, it owns slots and is an established cluster
// member even if cluster_known_nodes is 1 (e.g. a single-node cluster being
// scaled up); (2) a non-isolated pending node from a previous MEET batch;
// (3) the first isolated node as a bootstrap seed when every single node is
// isolated (fresh bootstrap, first reconcile).
func findMeetTarget(state *valkey.ClusterState, isolated []*valkey.NodeState) *valkey.NodeState {
	for _, shard := range state.Shards {
		if p := shard.GetPrimaryNode(); p != nil {
			return p
		}
	}
	for _, node := range state.PendingNodes {
		if !node.IsIsolated() {
			return node
		}
	}
	return isolated[0]
}

// meetIsolatedNodes issues CLUSTER MEET for every isolated pending node
// (cluster_known_nodes <= 1). Phase 2 (assignSlotsToPendingPrimaries)
// refuses to assign slots to isolated nodes, so every node is guaranteed
// to pass through this function before receiving slots or replicating.
//
// Before issuing MEET, we set each node's config epoch above the cluster's
// current epoch.
//
// MEET is idempotent and has no ordering dependencies, so all isolated nodes
// can be introduced in a single pass. We pick a single "meet target" node
// and MEET all others against it:
//
//   - If any non-isolated node exists (shard primary with cluster_known_nodes
//     > 1, or a pending node from a previous MEET batch), use it as the
//     target so new nodes join the existing cluster.
//   - If all nodes are isolated (fresh bootstrap), use the first isolated
//     node as a bootstrap seed. All others MEET this seed, and gossip
//     propagates the full topology from there.
//
// Returns the number of nodes that were MEET'd.
func (r *ValkeyClusterReconciler) meetIsolatedNodes(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState) (int, error) {
	log := logf.FromContext(ctx)

	var isolated []*valkey.NodeState
	for _, node := range state.PendingNodes {
		if node.IsIsolated() {
			isolated = append(isolated, node)
		}
	}
	if len(isolated) == 0 {
		return 0, nil
	}

	// Find a well-connected node to MEET all isolated nodes against.
	// Falls back to an isolated node as a bootstrap seed if every node
	// is isolated (fresh bootstrap).
	meetTarget := findMeetTarget(state, isolated)

	// Set config epochs on isolated nodes before MEET. Fresh nodes have
	// epoch 0; setting a higher epoch ensures they enter the cluster with
	// authority above any dead nodes they replace, preventing gossip from
	// overriding their future slot claims.
	currentEpoch := meetTarget.CurrentEpoch()
	for i, node := range isolated {
		epoch := currentEpoch + int64(i) + 1
		if err := node.Client.Do(ctx, node.Client.B().ClusterSetConfigEpoch().ConfigEpoch(epoch).Build()).Error(); err != nil {
			log.V(1).Info("CLUSTER SETCONFIGEPOCH skipped", "node", node.Address, "err", err)
		}
	}

	// Exclude the meet target from the MEET loop during bootstrap.
	if meetTarget == isolated[0] {
		isolated = isolated[1:]
	}

	met := 0
	for _, node := range isolated {
		// Bidirectional MEET: the isolated node MEETs the target, AND the
		// target MEETs the isolated node. Bidirectional MEET avoids the
		// fragmentation problem where one-way MEET + slow gossip creates
		// subclusters: by having the target actively reach out, the new
		// node is guaranteed to appear in the target's node table
		// immediately, not after gossip propagation.
		log.V(1).Info("meet node", "node", node.Address, "target", meetTarget.Address)
		if err := node.Client.Do(ctx, node.Client.B().ClusterMeet().Ip(meetTarget.Address).Port(int64(meetTarget.Port)).Build()).Error(); err != nil {
			log.Error(err, "CLUSTER MEET failed", "from", node.Address, "to", meetTarget.Address)
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "ClusterMeetFailed", "ClusterMeet", "CLUSTER MEET %v -> %v failed: %v", node.Address, meetTarget.Address, err)
			return met, err
		}
		if err := meetTarget.Client.Do(ctx, meetTarget.Client.B().ClusterMeet().Ip(node.Address).Port(int64(node.Port)).Build()).Error(); err != nil {
			log.Error(err, "CLUSTER MEET failed", "from", meetTarget.Address, "to", node.Address)
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "ClusterMeetFailed", "ClusterMeet", "CLUSTER MEET %v -> %v failed: %v", meetTarget.Address, node.Address, err)
			return met, err
		}
		met++
	}
	return met, nil
}

// assignSlotsToPendingPrimaries assigns hash-slot ranges to non-isolated
// pending nodes whose pod labels indicate they are primaries (node index 0
// within their shard). During scale-out (no unassigned slots available) it
// returns 0 — new primaries stay in PendingNodes for the rebalancer to handle.
// Isolated nodes (cluster_known_nodes <= 1) are skipped; they must be MEET'd
// first in Phase 1.
func (r *ValkeyClusterReconciler) assignSlotsToPendingPrimaries(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState, nodes *valkeyiov1alpha1.ValkeyNodeList) (int, error) {
	log := logf.FromContext(ctx)

	// Collect primary pending nodes (node index 0 = primary), skipping:
	//  - isolated nodes (cluster_known_nodes <= 1): they haven't been
	//    MEET'd yet and must go through Phase 1 first. Without this guard
	//    a single pod that starts before its peers would get slots while
	//    isolated, creating a disconnected shard primary.
	//    Exception: when the entire expected cluster is a single node
	//    (1 shard, 0 replicas), isolation is the permanent correct state,
	//    there is nobody to MEET, so the guard is skipped.
	//  - post-failover replacements whose shard already has a live primary.
	isSingleNodeCluster := cluster.Spec.Shards == 1 && cluster.Spec.Replicas == 0
	primaries := make([]*valkey.NodeState, 0, len(state.PendingNodes))
	for _, node := range state.PendingNodes {
		if node.IsIsolated() && !isSingleNodeCluster {
			continue
		}
		role, shardIndex := nodeRoleAndShard(node.Address, nodes)
		if role != RolePrimary {
			continue
		}
		if shardExistsInTopology(state, shardIndex, nodes) {
			log.V(1).Info("skipping slot assignment; shard already exists in topology (post-failover)",
				"shardIndex", shardIndex, "node", node.Address)
			continue
		}
		primaries = append(primaries, node)
	}
	if len(primaries) == 0 {
		return 0, nil
	}

	slots := state.GetUnassignedSlots()
	if len(slots) == 0 {
		// Scale-out: all slots already assigned to existing primaries.
		// New primaries stay in PendingNodes; the rebalancer will find
		// them and migrate slots to them.
		log.V(1).Info("no unassigned slots; new primaries will receive slots via rebalance", "count", len(primaries))
		return 0, nil
	}

	unassignedSlots := 0
	for _, s := range slots {
		unassignedSlots += s.End - s.Start + 1
	}

	assigned := 0
	for _, node := range primaries {
		if len(slots) == 0 {
			break
		}

		remainingPrimaries := len(primaries) - assigned
		// Ceiling division to ensure all slots are distributed without remainder.
		numSlotsForNode := (unassignedSlots + remainingPrimaries - 1) / remainingPrimaries

		// Collect ranges for this node
		var nodeRanges []valkey.SlotsRange
		remaining := numSlotsForNode
		for remaining > 0 && len(slots) > 0 {
			slotStart := slots[0].Start
			take := min(slots[0].End-slots[0].Start+1, remaining)
			slotEnd := slotStart + take - 1

			nodeRanges = append(nodeRanges, valkey.SlotsRange{Start: slotStart, End: slotEnd})
			remaining -= take
			unassignedSlots -= take
			// Advance the slots list: either consume the entire first range
			// or trim its start. This avoids rebuilding the slice each iteration.
			if take == slots[0].End-slots[0].Start+1 {
				slots = slots[1:]
			} else {
				slots[0].Start = slotEnd + 1
			}
		}

		// Issue a single CLUSTER ADDSLOTSRANGE with all ranges for this node
		cmd := node.Client.B().ClusterAddslotsrange().StartSlotEndSlot()
		for _, sr := range nodeRanges {
			cmd = cmd.StartSlotEndSlot(int64(sr.Start), int64(sr.End))
		}
		log.V(1).Info("assign slots to primary", "node", node.Address, "ranges", nodeRanges)
		if err := node.Client.Do(ctx, cmd.Build()).Error(); err != nil {
			log.Error(err, "CLUSTER ADDSLOTSRANGE failed", "node", node.Address, "ranges", nodeRanges)
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "SlotAssignmentFailed", "AssignSlots", "Failed to assign slots %s to %v: %v", valkey.FormatSlotsRanges(nodeRanges), node.Address, err)
			return assigned, err
		}

		r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "PrimaryCreated", "CreatePrimary", "Created primary %v with slots %s", node.Address, valkey.FormatSlotsRanges(nodeRanges))
		assigned++
	}
	return assigned, nil
}

// errPrimaryNotReady is returned by replicateToShardPrimary when the
// primary for a shard hasn't received slots yet (e.g. during scale-out,
// while the rebalancer is still migrating slots), or when gossip hasn't
// propagated the primary's node ID to the replica yet. This is not a
// fatal error — the replica will be retried on a future reconcile.
var errPrimaryNotReady = errors.New("primary not yet in cluster state (awaiting rebalance)")

// replicatePendingReplicas issues CLUSTER REPLICATE for all pending nodes
// whose pod labels indicate they are replicas (node index 1+), as well as
// post-failover replacement primaries (node index 0 whose shard already has
// a live primary). During initial creation all primaries should already
// have slots, but during scale-out new primaries may still be empty while the
// rebalancer migrates slots to them. Replicas whose primary isn't ready yet
// are silently skipped and retried on the next reconcile.
// Returns the number of replicas that were attached.
func (r *ValkeyClusterReconciler) replicatePendingReplicas(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState, nodes *valkeyiov1alpha1.ValkeyNodeList) (int, error) {
	log := logf.FromContext(ctx)
	replicated := 0

	for _, node := range state.PendingNodes {
		role, shardIndex := nodeRoleAndShard(node.Address, nodes)
		// Normal replicas (node-index >= 1) always need replication.
		// Post-failover replacements (node-index=0 but shard already has
		// a live primary) also need replication — they were skipped by
		// assignSlotsToPendingPrimaries.
		if role == RolePrimary {
			if !shardExistsInTopology(state, shardIndex, nodes) {
				continue // genuine new primary, handled by assignSlotsToPendingPrimaries
			}
			log.V(1).Info("post-failover: attaching replacement node as replica",
				"shardIndex", shardIndex, "node", node.Address)
		}

		if err := r.replicateToShardPrimary(ctx, cluster, state, node, shardIndex, nodes); err != nil {
			if errors.Is(err, errPrimaryNotReady) {
				log.V(1).Info("skipping replica; primary not ready yet", "node", node.Address, "shard", shardIndex)
				continue
			}
			log.Error(err, "failed to replicate pending node", "node", node.Address, "shard", shardIndex)
			return replicated, err
		}
		replicated++
	}
	return replicated, nil
}

// replicateToShardPrimary issues CLUSTER REPLICATE to attach this node as a
// replica of the primary in the same shard.
//
// The primary is found by scanning all pods in the shard against the live
// Valkey topology (via findShardPrimary). This handles both the normal case
// (node-index=0 is the primary) and the post-failover case (a promoted
// replica is the primary). If no primary is found (e.g. the primary pod
// hasn't started yet or hasn't joined the cluster), we return an error and
// the reconciler retries on the next cycle.
func (r *ValkeyClusterReconciler) replicateToShardPrimary(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState, node *valkey.NodeState, shardIndex int, nodes *valkeyiov1alpha1.ValkeyNodeList) error {
	log := logf.FromContext(ctx)

	// Find the actual primary for this shard by scanning all shard pods
	// against the live Valkey topology. This handles both the normal case
	// (node-index=0 is the primary) and the post-failover case (a promoted
	// replica is the primary).
	primaryNodeId, primaryIP := findShardPrimary(state, shardIndex, nodes)
	if primaryNodeId == "" {
		return fmt.Errorf("shard %d: %w", shardIndex, errPrimaryNotReady)
	}

	log.V(1).Info("add a new replica", "primary IP", primaryIP, "primary Id", primaryNodeId, "replica address", node.Address, "shardIndex", shardIndex)
	if err := node.Client.Do(ctx, node.Client.B().ClusterReplicate().NodeId(primaryNodeId).Build()).Error(); err != nil {
		// "Unknown node" means gossip hasn't propagated the primary's ID to
		// this replica yet. This is transient and will resolve on the next
		// reconcile once gossip catches up — treat it as retriable.
		if strings.Contains(err.Error(), "Unknown node") {
			log.V(1).Info("replica does not yet know primary (gossip pending); will retry", "replica", node.Address, "primaryId", primaryNodeId)
			return fmt.Errorf("shard %d: %w", shardIndex, errPrimaryNotReady)
		}
		log.Error(err, "command failed: CLUSTER REPLICATE", "nodeId", primaryNodeId)
		r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "ReplicaCreationFailed", "CreateReplica", "Failed to create replica: %v", err)
		return err
	}
	r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ReplicaCreated", "CreateReplica", "Created replica for primary %v (shard %d)", primaryNodeId, shardIndex)
	return nil
}

// promoteOrphanedReplicas issues CLUSTER FAILOVER TAKEOVER to replicas whose
// primary is dead and failover quorum is unreachable. With persistence enabled
// the pod will return with the same node ID, so TAKEOVER is skipped.
// Returns (result, true) if a promotion was made and reconcile should requeue.
func (r *ValkeyClusterReconciler) promoteOrphanedReplicas(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState) (ctrl.Result, bool) {
	if cluster.Spec.Persistence != nil || state.HasFailoverQuorum() {
		return ctrl.Result{}, false
	}
	log := logf.FromContext(ctx)
	var promoted, attempted int
	deadPrimaries := make(map[string]bool)
	for _, shard := range state.Shards {
		for _, node := range shard.Nodes {
			primaryId := node.PrimaryIdFromSelf()
			if primaryId == "-" || primaryId == "" {
				continue // node is a primary
			}
			if deadPrimaries[primaryId] {
				continue
			}
			if !state.IsNodeFailed(primaryId) {
				continue
			}
			// Note: we don't check if the pod still exists in k8s here.
			// Without persistence, even if the primary is merely partitioned
			// (not gone), the TAKEOVER'd replica will own the slots at a
			// higher epoch. A returning primary will see the higher epoch
			// and demote itself; no split-brain.
			deadPrimaries[primaryId] = true
		}
	}
	for primaryId := range deadPrimaries {
		replica := state.BestReplicaOf(primaryId)
		if replica == nil {
			continue
		}
		attempted++
		log.Info("promoting orphaned replica via TAKEOVER", "node", replica.Address, "deadPrimary", primaryId)
		if err := replica.Client.Do(ctx, replica.Client.B().ClusterFailover().Takeover().Build()).Error(); err != nil {
			log.Error(err, "CLUSTER FAILOVER TAKEOVER failed", "node", replica.Address)
		} else {
			promoted++
		}
	}
	if promoted > 0 || attempted > 0 {
		if promoted > 0 {
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ReplicasTakenOver", "FailoverTakeover", "Promoted %d orphaned replica(s) via TAKEOVER", promoted)
		}
		return ctrl.Result{RequeueAfter: 2 * time.Second}, true
	}
	return ctrl.Result{}, false
}

// Check each cluster node and forget stale nodes (noaddr or status fail)
func (r *ValkeyClusterReconciler) forgetStaleNodes(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState, nodes *valkeyiov1alpha1.ValkeyNodeList) {
	log := logf.FromContext(ctx)
	for _, shard := range state.Shards {
		for _, node := range shard.Nodes {
			for _, failing := range node.GetFailingNodes() {
				idx := slices.IndexFunc(nodes.Items, func(n valkeyiov1alpha1.ValkeyNode) bool {
					return n.Status.PodIP == failing.Address
				})
				if idx != -1 {
					continue
				}
				// A live replica still considers this failing node its
				// primary. Forgetting it from the other primaries now would
				// remove it from their node tables and prevent them from
				// voting in the auto-failover election, permanently
				// blocking the replica's promotion.
				if state.HasReplicaOf(failing.Id) {
					if cluster.Spec.Persistence != nil || state.HasFailoverQuorum() {
						log.V(1).Info("skipping forget; failover pending for node",
							"address", failing.Address, "Id", failing.Id)
						continue
					}
					log.Info("forget node despite pending replica; quorum unreachable",
						"address", failing.Address, "Id", failing.Id)
				} else {
					log.V(1).Info("forget a failing node", "address", failing.Address, "Id", failing.Id)
				}
				if err := node.Client.Do(ctx, node.Client.B().ClusterForget().NodeId(failing.Id).Build()).Error(); err != nil {
					log.Error(err, "command failed: CLUSTER FORGET")
					r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "NodeForgetFailed", "ForgetNode", "Failed to forget node: %v", err)
				} else {
					r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "StaleNodeForgotten", "ForgetNode", "Forgot stale node %v", failing.Address)
				}
			}
		}
	}
}

// updateStatus updates the status with the current conditions and computes the Valkey Cluster state
func (r *ValkeyClusterReconciler) updateStatus(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState) error {
	log := logf.FromContext(ctx)
	// Fetch the latest version to avoid conflicts from stale resourceVersion
	current := &valkeyiov1alpha1.ValkeyCluster{}
	if err := r.Get(ctx, client.ObjectKeyFromObject(cluster), current); err != nil {
		return err
	}

	// Snapshot for patch base and comparison
	patchBase := current.DeepCopy()
	patch := client.MergeFrom(patchBase)

	// Update shard counts
	if state != nil {
		current.Status.ReadyShards = r.countReadyShards(state, cluster)
		current.Status.Shards = int32(len(state.Shards))
	}

	// Apply conditions from the in-memory cluster object
	current.Status.Conditions = cluster.Status.Conditions

	// compute Valkey Cluster state from conditions (priority order: Degraded > Ready > Progressing > Failed)
	readyCondition := meta.FindStatusCondition(current.Status.Conditions, valkeyiov1alpha1.ConditionReady)
	progressingCondition := meta.FindStatusCondition(current.Status.Conditions, valkeyiov1alpha1.ConditionProgressing)
	degradedCondition := meta.FindStatusCondition(current.Status.Conditions, valkeyiov1alpha1.ConditionDegraded)

	switch {
	case degradedCondition != nil && degradedCondition.Status == metav1.ConditionTrue:
		current.Status.State = valkeyiov1alpha1.ClusterStateDegraded
		current.Status.Reason = degradedCondition.Reason
		current.Status.Message = degradedCondition.Message
	case readyCondition != nil && readyCondition.Status == metav1.ConditionTrue:
		current.Status.State = valkeyiov1alpha1.ClusterStateReady
		current.Status.Reason = readyCondition.Reason
		current.Status.Message = readyCondition.Message
	case progressingCondition != nil && progressingCondition.Status == metav1.ConditionTrue:
		current.Status.State = valkeyiov1alpha1.ClusterStateReconciling
		current.Status.Reason = progressingCondition.Reason
		current.Status.Message = progressingCondition.Message
	case readyCondition != nil && readyCondition.Status == metav1.ConditionFalse:
		current.Status.State = valkeyiov1alpha1.ClusterStateFailed
		current.Status.Reason = readyCondition.Reason
		current.Status.Message = readyCondition.Message
	}

	// Update Prometheus metrics
	updateClusterMetrics(current)

	if !reflect.DeepEqual(patchBase.Status, current.Status) {
		if err := r.Status().Patch(ctx, current, patch); err != nil {
			log.Error(err, statusUpdateFailedMsg)
			return err
		}
		log.V(1).Info("status updated", "state", current.Status.State, "reason", current.Status.Reason)
	} else {
		log.V(2).Info("status unchanged, skipping update")
	}

	// Sync back to caller so subsequent logic sees the latest state
	cluster.Status = current.Status
	cluster.ResourceVersion = current.ResourceVersion
	return nil
}

// countReadyShards counts shards that have all required nodes, are healthy,
// and have all replicas in sync with their primary.
func (r *ValkeyClusterReconciler) countReadyShards(state *valkey.ClusterState, cluster *valkeyiov1alpha1.ValkeyCluster) int32 {
	var readyCount int32 = 0
	requiredNodes := 1 + int(cluster.Spec.Replicas)
	for _, shard := range state.Shards {
		if len(shard.Nodes) < requiredNodes || shard.GetPrimaryNode() == nil {
			continue
		}
		// Check if all nodes in this shard are healthy and in sync
		allHealthy := true
		for _, node := range shard.Nodes {
			if slices.Contains(node.Flags, "fail") || slices.Contains(node.Flags, "pfail") {
				allHealthy = false
				break
			}
			if !node.IsReplicationInSync() {
				allHealthy = false
				break
			}
		}
		if allHealthy {
			readyCount++
		}
	}
	return readyCount
}

const rebalanceSlotBatchSize = 400

func (r *ValkeyClusterReconciler) rebalanceSlots(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, shards []*valkey.ShardState) (bool, error) {
	move, err := valkey.PlanRebalanceMove(shards, int(cluster.Spec.Shards), rebalanceSlotBatchSize)
	if err != nil {
		return false, err
	}
	if move == nil {
		return false, nil
	}

	log := logf.FromContext(ctx)
	inProgress, err := valkey.SlotMigrationInProgress(ctx, move.Src)
	if err != nil {
		return false, err
	}
	if inProgress {
		log.V(1).Info("slot migration already in progress", "src", move.Src.Address)
		return true, nil
	}

	if !strings.Contains(move.Src.ClusterNodes, move.Dst.Id) {
		log.V(1).Info("destination not yet visible to source via gossip; will retry", "src", move.Src.Address, "dst", move.Dst.Address, "dstId", move.Dst.Id)
		r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "SlotsRebalancePending", "RebalanceSlots", "Waiting for %s to learn node %s", move.Src.Address, move.Dst.Address)
		return true, nil
	}

	log.V(1).Info("rebalancing slots", "src", move.Src.Address, "dst", move.Dst.Address, "slots", len(move.Slots))
	r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "SlotsRebalancing", "RebalanceSlots", "Moving %d slots from %s to %s", len(move.Slots), move.Src.Address, move.Dst.Address)

	ranges := valkey.SlotsToRanges(move.Slots)
	if err := valkey.MigrateSlotsAtomic(ctx, move.Src, move.Dst, ranges); err != nil {
		if valkey.IsSlotsNotServedByNode(err) {
			log.V(1).Info("slots no longer served by source; will retry with fresh state", "src", move.Src.Address, "dst", move.Dst.Address)
			return true, nil
		}
		return false, err
	}
	slotMigrationBatchesTotal.WithLabelValues(cluster.Name, cluster.Namespace).Inc()
	return true, nil
}

// effectiveShards returns state.Shards plus any pending primaries that are
// scale-out leaders (slot-less Valkey primaries labeled as RolePrimary).
func effectiveShards(state *valkey.ClusterState, nodes *valkeyiov1alpha1.ValkeyNodeList) []*valkey.ShardState {
	shards := append([]*valkey.ShardState(nil), state.Shards...)
	for _, node := range state.PendingNodes {
		role, _ := nodeRoleAndShard(node.Address, nodes)
		if role == RolePrimary {
			shards = append(shards, &valkey.ShardState{
				Id:        node.ShardId,
				Nodes:     []*valkey.NodeState{node},
				PrimaryId: node.Id,
			})
		}
	}
	return shards
}

// handleScaleIn drains excess shards and cleans up leftover ValkeyNodes.
// Returns (result, true) when work was done or an error occurred and the
// caller should requeue instead of continuing to health checks.
func (r *ValkeyClusterReconciler) handleScaleIn(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState, nodes *valkeyiov1alpha1.ValkeyNodeList) (ctrl.Result, bool) {
	log := logf.FromContext(ctx)

	if len(state.Shards) > int(cluster.Spec.Shards) {
		drained, err := r.drainExcessShards(ctx, cluster, state, nodes)
		if err != nil {
			log.Error(err, "scale-in draining failed")
			r.Recorder.Eventf(cluster, nil, corev1.EventTypeWarning, "DrainFailed", "ScaleIn", "Failed to drain excess shards: %v", err)
			setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonRebalanceFailed, err.Error(), metav1.ConditionTrue)
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Scaling in cluster", metav1.ConditionFalse)
			setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonRebalancingSlots, "Rebalancing slots for scale-in", metav1.ConditionTrue)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, true
		}
		if drained {
			meta.RemoveStatusCondition(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionDegraded)
			setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonReconciling, "Scaling in cluster", metav1.ConditionFalse)
			setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonRebalancingSlots, "Rebalancing slots for scale-in", metav1.ConditionTrue)
			_ = r.updateStatus(ctx, cluster, state)
			return ctrl.Result{RequeueAfter: 2 * time.Second}, true
		}
	}

	// Clean up leftover ValkeyNodes from a previous scale-in where drained
	// primaries became replicas before their ValkeyNodes could be deleted.
	if deleted, err := r.deleteExcessValkeyNodes(ctx, cluster); err != nil {
		log.Error(err, "failed to delete excess ValkeyNodes")
		return ctrl.Result{RequeueAfter: 2 * time.Second}, true
	} else if deleted {
		return ctrl.Result{RequeueAfter: 2 * time.Second}, true
	}

	return ctrl.Result{}, false
}

// drainExcessShards handles scale-in by migrating slots away from shards
// whose pod shard-index >= spec.Shards. Once a shard is fully drained (0
// slots), its ValkeyNodes are deleted; forgetStaleNodes on the next reconcile
// cleans up the Valkey topology.
// Returns true if any work was done (caller should requeue).
func (r *ValkeyClusterReconciler) drainExcessShards(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, state *valkey.ClusterState, nodes *valkeyiov1alpha1.ValkeyNodeList) (bool, error) {
	log := logf.FromContext(ctx)
	expectedShards := int(cluster.Spec.Shards)

	var remaining, draining []*valkey.ShardState
	for _, shard := range state.Shards {
		shardIndex := shardIndexFromState(shard, nodes)
		if shardIndex >= 0 && shardIndex < expectedShards {
			remaining = append(remaining, shard)
		} else {
			draining = append(draining, shard)
		}
	}
	if len(draining) == 0 {
		return false, nil
	}

	for _, shard := range draining {
		move, err := valkey.PlanDrainMove(shard, remaining, rebalanceSlotBatchSize)
		if err != nil {
			return false, err
		}
		if move == nil {
			continue
		}

		inProgress, err := valkey.SlotMigrationInProgress(ctx, move.Src)
		if err != nil {
			return false, err
		}
		if inProgress {
			log.V(1).Info("drain migration in progress", "src", move.Src.Address)
			return true, nil
		}

		if !strings.Contains(move.Src.ClusterNodes, move.Dst.Id) {
			log.V(1).Info("drain destination not yet known to source", "src", move.Src.Address, "dst", move.Dst.Address)
			return true, nil
		}

		log.V(1).Info("draining slots", "src", move.Src.Address, "dst", move.Dst.Address, "slots", len(move.Slots))
		r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "SlotsDraining", "ScaleIn", "Moving %d slots from %s to %s", len(move.Slots), move.Src.Address, move.Dst.Address)

		ranges := valkey.SlotsToRanges(move.Slots)
		if err := valkey.MigrateSlotsAtomic(ctx, move.Src, move.Dst, ranges); err != nil {
			if valkey.IsSlotsNotServedByNode(err) {
				return true, nil
			}
			return false, err
		}
		slotMigrationBatchesTotal.WithLabelValues(cluster.Name, cluster.Namespace).Inc()
		return true, nil
	}

	// All draining shards have 0 slots — delete their ValkeyNodes.
	for _, shard := range draining {
		shardIndex := shardIndexFromState(shard, nodes)
		if shardIndex < 0 {
			continue
		}
		nodesPerShard := 1 + int(cluster.Spec.Replicas)
		for nodeIndex := range nodesPerShard {
			name := valkeyNodeName(cluster.Name, shardIndex, nodeIndex)
			node := &valkeyiov1alpha1.ValkeyNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      name,
					Namespace: cluster.Namespace,
				},
			}
			if err := r.Delete(ctx, node); err != nil {
				if !apierrors.IsNotFound(err) {
					return false, fmt.Errorf("delete ValkeyNode %s: %w", name, err)
				}
			} else {
				log.V(1).Info("deleted ValkeyNode for drained shard", "name", name, "shard", shardIndex)
				r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ValkeyNodeDeleted", "ScaleIn", "Deleted ValkeyNode %s (shard %d)", name, shardIndex)
			}
		}
	}
	return true, nil
}

// deleteExcessValkeyNodes removes ValkeyNode CRs that are outside the desired
// spec: shard-index >= spec.Shards OR node-index >= 1 + spec.Replicas.
func (r *ValkeyClusterReconciler) deleteExcessValkeyNodes(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster) (bool, error) {
	log := logf.FromContext(ctx)
	allNodes := &valkeyiov1alpha1.ValkeyNodeList{}
	if err := r.List(ctx, allNodes, client.InNamespace(cluster.Namespace), client.MatchingLabels(map[string]string{LabelCluster: cluster.Name})); err != nil {
		return false, err
	}
	nodesPerShard := 1 + int(cluster.Spec.Replicas)
	deleted := false
	for i := range allNodes.Items {
		node := &allNodes.Items[i]
		shardIndex, err := strconv.Atoi(node.Labels[LabelShardIndex])
		if err != nil {
			continue
		}
		nodeIndex, err := strconv.Atoi(node.Labels[LabelNodeIndex])
		if err != nil {
			continue
		}
		if shardIndex >= int(cluster.Spec.Shards) || nodeIndex >= nodesPerShard {
			if err := r.Delete(ctx, node); err != nil {
				if !apierrors.IsNotFound(err) {
					return false, fmt.Errorf("delete excess ValkeyNode %s: %w", node.Name, err)
				}
			} else {
				log.V(1).Info("deleted excess ValkeyNode", "name", node.Name, "shard", shardIndex, "node", nodeIndex)
				r.Recorder.Eventf(cluster, nil, corev1.EventTypeNormal, "ValkeyNodeDeleted", "ScaleIn", "Deleted excess ValkeyNode %s (shard %d, node %d)", node.Name, shardIndex, nodeIndex)
				deleted = true
			}
		}
	}
	return deleted, nil
}

// shardIndexFromState determines the shard index for a given Valkey Cluster
// shard by matching its node addresses to ValkeyNode shard-index labels. It checks
// the primary first, falling back to any node if unavailable.
func shardIndexFromState(shard *valkey.ShardState, nodes *valkeyiov1alpha1.ValkeyNodeList) int {
	if primary := shard.GetPrimaryNode(); primary != nil {
		_, shardIndex := nodeRoleAndShard(primary.Address, nodes)
		if shardIndex >= 0 {
			return shardIndex
		}
	}
	for _, node := range shard.Nodes {
		_, shardIndex := nodeRoleAndShard(node.Address, nodes)
		if shardIndex >= 0 {
			return shardIndex
		}
	}
	return -1
}

// SetupWithManager sets up the controller with the Manager.
func (r *ValkeyClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	r.APIReader = mgr.GetAPIReader()
	return ctrl.NewControllerManagedBy(mgr).
		For(&valkeyiov1alpha1.ValkeyCluster{}).
		Owns(&corev1.Service{}).
		Owns(&corev1.ConfigMap{}).
		Owns(&valkeyiov1alpha1.ValkeyNode{}).
		Owns(&corev1.Secret{}).
		Owns(&policyv1.PodDisruptionBudget{}).
		Named("valkeycluster").
		Complete(r)
}