fix: self-heal partial-slot outage (slots stuck on failed master)

internal/controller/cluster_helpers_test.go

@@ -261,8 +261,13 @@ func TestDecidePhase_matrix(t *testing.T) {
 		vc.Status.Version = status
 		return vc
 	}
+	// 기본적으로 slots_ok == slots_assigned (healthy). partial-outage 케이스는
+	// mkInfoOK 로 별도 구성.
 	mkInfo := func(state string, slots int32) *vk.ClusterInfo {
-		return &vk.ClusterInfo{State: state, SlotsAssigned: slots}
+		return &vk.ClusterInfo{State: state, SlotsAssigned: slots, SlotsOK: slots}
+	}
+	mkInfoOK := func(state string, slots, slotsOK int32) *vk.ClusterInfo {
+		return &vk.ClusterInfo{State: state, SlotsAssigned: slots, SlotsOK: slotsOK}
 	}
 	cases := []struct {
 		name         string
@@ -278,6 +283,8 @@ func TestDecidePhase_matrix(t *testing.T) {
 		{"sts ready, cluster not ok", mkVC("8.1.6", "8.1.6"), 6, 6, mkInfo("fail", 0), cachev1alpha1.ClusterPhaseInitializing},
 		{"resharding", mkVC("8.1.6", "8.1.6"), 6, 6, mkInfo("ok", 8192), cachev1alpha1.ClusterPhaseResharding},
 		{"running", mkVC("8.1.6", "8.1.6"), 6, 6, mkInfo("ok", 16384), cachev1alpha1.ClusterPhaseRunning},
+		// 결함 ⑤: state=ok + assigned=16384 이지만 slots_ok<16384 → Running 아님(Resharding cadence).
+		{"partial-slot outage", mkVC("8.1.6", "8.1.6"), 6, 6, mkInfoOK("ok", 16384, 10922), cachev1alpha1.ClusterPhaseResharding},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {

internal/controller/cluster_stuckslots.go

@@ -0,0 +1,122 @@
+/*
+Copyright 2026 Keiailab.
+
+Licensed under the MIT License. See the LICENSE file for details.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	commonsevents "github.com/keiailab/keiailab-commons/pkg/events"
+	cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1"
+	"github.com/keiailab/valkey-operator/internal/observability"
+	vk "github.com/keiailab/valkey-operator/internal/valkey"
+)
+
+// 결함 ⑤ — partial-slot outage 자가복구.
+//
+// 증상: node churn 후 cluster_state:ok + slots_assigned:16384 이지만 slots_ok 가
+// 16384 미만 — 한 shard 의 slot 이 `fail` flag 가 붙은 master 소유로 남아 그만큼의
+// 키스페이스가 DOWN. 기존 health gate (cluster_state / slots_assigned) 는 이를
+// "정상" 으로 오판해 self-heal 이 멈췄다.
+//
+// 본 path 는:
+//  1. vk.IsClusterDegraded: slots_ok<16384 또는 fail master 가 slot 소유 시 degraded.
+//  2. vk.DetectStuckSlotHeals: 각 stuck master 마다 takeover 가능한 healthy replica 를
+//     선정 (순수 함수, 단위테스트).
+//  3. healStuckSlots: 선정된 replica 에 CLUSTER FAILOVER TAKEOVER 발행 → slot 소유권이
+//     replica 로 승계되어 slot 이 ok 로 복귀.
+//
+// 보수성 / 멱등:
+//   - `fail` flag 는 Valkey 가 node-timeout 경과 + gossip 합의 후에만 설정한다. 따라서
+//     flag 의 존재 자체가 "node-timeout 만큼 기다렸다" 는 신호 — converging cluster 를
+//     thrash 하지 않는다 (pfail/fail? 는 takeover 대상으로 보지 않는다).
+//   - takeover 후 master 가 다시 healthy 가 되면 stuck 조건이 사라져 자동 no-op.
+//   - healthy replica 가 없는 stuck master 는 TAKEOVER 하지 않고 (데이터 보존) 경고만.
+
+// reconcileStuckSlots — partial-slot outage 게이트 + 복구. allReady && cluster 도달
+// 가능한 상태에서 호출. 반환값은 발행한 takeover 수 (0 이면 no-op).
+func (r *ValkeyClusterReconciler) reconcileStuckSlots(
+	ctx context.Context, vc *cachev1alpha1.ValkeyCluster,
+	password string, info *vk.ClusterInfo, nodes []vk.NodeView,
+) (int, error) {
+	if info == nil || len(nodes) == 0 {
+		return 0, nil
+	}
+	if !vk.IsClusterDegraded(info, nodes) {
+		return 0, nil
+	}
+
+	logger := log.FromContext(ctx)
+	heals := vk.DetectStuckSlotHeals(nodes)
+	if len(heals) == 0 {
+		// degraded 이지만 안전한 takeover 후보가 없다 (예: fail master 에 healthy
+		// replica 부재, 또는 slots_ok 저하가 pfail 같은 transient 상태). 이번 reconcile
+		// 에서는 행동하지 않고 가시화만 — 다음 reconcile 에서 gossip 수렴 또는 멤버십
+		// 자가복구(결함 ③)가 replica 를 재합류시킨 뒤 재평가한다.
+		logger.Info("ValkeyCluster degraded (partial-slot outage) but no safe takeover candidate",
+			"slotsOK", info.SlotsOK, "slotsAssigned", info.SlotsAssigned, "state", info.State)
+		commonsevents.EmitWarningf(r.Recorder, vc, "PartialSlotOutage",
+			"slots_ok=%d/%d — fail master 의 slot 이 stuck 이나 takeover 가능한 healthy replica 없음",
+			info.SlotsOK, vk.ClusterTotalSlots)
+		return 0, nil
+	}
+
+	logger.Info("ValkeyCluster partial-slot outage detected; healing via TAKEOVER",
+		"slotsOK", info.SlotsOK, "stuckShards", len(heals))
+
+	return r.healStuckSlots(ctx, vc, password, heals)
+}
+
+// healStuckSlots — 각 heal 계획의 replica 에 CLUSTER FAILOVER TAKEOVER 발행.
+//
+// TAKEOVER 는 합의 없이 replica 가 slot + epoch 를 승계 — master 가 이미 fail 일 때만
+// 안전하며 (DetectStuckSlotHeals 가 그것을 보장), split-brain 위험 때문에 그 외에는
+// 호출하지 않는다.
+func (r *ValkeyClusterReconciler) healStuckSlots(
+	ctx context.Context, vc *cachev1alpha1.ValkeyCluster,
+	password string, heals []vk.StuckSlotHeal,
+) (int, error) {
+	ctx, span := observability.StartCallSpan(ctx, "ValkeyCluster/HealStuckSlots")
+	defer span.End()
+	logger := log.FromContext(ctx)
+
+	tlsCfg, err := r.tlsConfigForCluster(ctx, vc)
+	if err != nil {
+		return 0, fmt.Errorf("tls config: %w", err)
+	}
+
+	var done int
+	var firstErr error
+	for _, h := range heals {
+		c := dialPod(h.TakeoverReplicaAddr, password, tlsCfg)
+		takeoverCtx, takeoverSpan := observability.StartCallSpan(ctx, "ValkeyCluster/FailoverTakeover")
+		takeErr := vk.ClusterFailoverTakeover(takeoverCtx, c)
+		if takeErr != nil {
+			takeoverSpan.RecordError(takeErr)
+		}
+		takeoverSpan.End()
+		_ = c.Close()
+		if takeErr != nil {
+			logger.Error(takeErr, "CLUSTER FAILOVER TAKEOVER failed — will retry next reconcile",
+				"replica", h.TakeoverReplicaAddr, "failedMaster", h.FailedMasterAddr)
+			if firstErr == nil {
+				firstErr = takeErr
+			}
+			continue
+		}
+		done++
+		MetricStuckSlotTakeoverTotal.WithLabelValues(vc.Namespace, vc.Name).Inc()
+		logger.Info("Stuck-slot takeover issued",
+			"replica", h.TakeoverReplicaAddr, "promotedFromFailedMaster", h.FailedMasterAddr)
+		commonsevents.Emitf(r.Recorder, vc, "StuckSlotTakeover",
+			"partial-slot outage 자가복구: replica %s 가 fail master %s 의 slot 승계 (CLUSTER FAILOVER TAKEOVER)",
+			h.TakeoverReplicaAddr, h.FailedMasterAddr)
+	}
+	return done, firstErr
+}

internal/controller/conditions_test.go

@@ -29,7 +29,7 @@ func TestApplyClusterConditions_running_state(t *testing.T) {
 	vc.Generation = 5
 	vc.Status.Phase = cachev1alpha1.ClusterPhaseRunning
 	conds := []metav1.Condition{}
-	info := &vk.ClusterInfo{State: "ok", SlotsAssigned: 16384}
+	info := &vk.ClusterInfo{State: "ok", SlotsAssigned: 16384, SlotsOK: 16384}
 
 	applyClusterConditions(&conds, vc, info, 6, 6)
 
@@ -96,6 +96,25 @@ func TestApplyClusterConditions_resharding(t *testing.T) {
 	}
 }
 
+// 결함 ⑤ — partial-slot outage 는 ClusterReady=False (Reason=PartialSlotOutage).
+func TestApplyClusterConditions_partialSlotOutage(t *testing.T) {
+	vc := &cachev1alpha1.ValkeyCluster{}
+	vc.Status.Phase = cachev1alpha1.ClusterPhaseResharding
+	conds := []metav1.Condition{}
+	// state=ok + assigned=16384 이지만 slots_ok<16384.
+	info := &vk.ClusterInfo{State: "ok", SlotsAssigned: 16384, SlotsOK: 10922}
+
+	applyClusterConditions(&conds, vc, info, 6, 6)
+
+	c := findCondition(conds, CondTypeClusterReady)
+	if c.Status != metav1.ConditionFalse {
+		t.Errorf("ClusterReady during partial outage must be False: %v", c.Status)
+	}
+	if c.Reason != "PartialSlotOutage" {
+		t.Errorf("Reason: %q want PartialSlotOutage", c.Reason)
+	}
+}
+
 func TestApplyClusterConditions_scalePending(t *testing.T) {
 	vc := &cachev1alpha1.ValkeyCluster{}
 	vc.Status.Phase = cachev1alpha1.ClusterPhaseRunning
@@ -106,7 +125,7 @@ func TestApplyClusterConditions_scalePending(t *testing.T) {
 	}
 	conds := []metav1.Condition{}
 
-	applyClusterConditions(&conds, vc, &vk.ClusterInfo{State: "ok", SlotsAssigned: 16384}, 6, 6)
+	applyClusterConditions(&conds, vc, &vk.ClusterInfo{State: "ok", SlotsAssigned: 16384, SlotsOK: 16384}, 6, 6)
 
 	c := findCondition(conds, CondTypeScalePending)
 	if c.Status != metav1.ConditionTrue {

internal/controller/metrics.go

@@ -129,6 +129,18 @@ var (
 		labelNamespaceName,
 	)
 
+	// MetricStuckSlotTakeoverTotal — 결함 ⑤ partial-slot outage 자가복구로 발행한
+	// CLUSTER FAILOVER TAKEOVER 카운터. fail master 의 slot 이 healthy replica 로
+	// 승계될 때 (성공 분기) 증가.
+	MetricStuckSlotTakeoverTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: metricSubsystem,
+			Name:      "stuck_slot_takeover_total",
+			Help:      "Total CLUSTER FAILOVER TAKEOVER events to heal partial-slot outages (Cluster mode)",
+		},
+		labelNamespaceName,
+	)
+
 	// MetricCapabilityActive — CR 의 활성 optional capability 추적 (PR #62 Status.Capabilities
 	// 의 Prometheus 측 노출). fleet-wide 채택 추적 용:
 	//   sum by (capability) (valkey_cluster_capability_active) → namespace 별 채택 CR 수
@@ -169,6 +181,7 @@ func init() {
 		MetricBackupTotal,
 		MetricRestoreTotal,
 		MetricFailoverTotal,
+		MetricStuckSlotTakeoverTotal,
 		MetricCapabilityActive,
 		MetricBuildInfo,
 	)

internal/controller/valkeycluster_controller.go

@@ -397,6 +397,25 @@ func (r *ValkeyClusterReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		}
 	}
 
+	// 10.6 partial-slot outage 자가복구 (결함 ⑤) — slot 레벨 health gate (cluster_state
+	//      =ok && slots_assigned=16384) 를 통과해도 *slots_ok<16384 또는 fail master 가
+	//      slot 을 소유* 하면 키스페이스 일부가 DOWN 이다. node churn 후 fail master 의
+	//      slot 이 stuck 인 경우, 그 master 의 healthy replica 에 CLUSTER FAILOVER TAKEOVER
+	//      를 발행해 slot 소유권을 승계시킨다. 보수적: `fail` flag 는 node-timeout 경과
+	//      후에만 설정되므로 converging cluster 를 thrash 하지 않으며, 멱등(복구되면 no-op).
+	if allReady && info != nil {
+		if healed, hErr := r.reconcileStuckSlots(ctx, vc, password, info, nodes); hErr != nil {
+			logger.Error(hErr, "Partial-slot outage heal pending — will retry")
+		} else if healed > 0 {
+			logger.Info("Healed partial-slot outage via takeover", "count", healed)
+			// takeover 직후 NODES/INFO 를 다시 읽어 status 정확도를 높인다 (slot 소유권
+			// 승계가 막 일어났다).
+			if i2, n2, e2 := r.pollClusterState(ctx, vc, password); e2 == nil && i2 != nil {
+				info, nodes = i2, n2
+			}
+		}
+	}
+
 	// 11. Shard status 빌드 + metrics 갱신.
 	//     ADR-0004 후속: NODES 응답이 있으면 *실제 토폴로지* 기반 — failover 정확.
 	//     없으면 spec 기반 fallback (부트스트랩 직후 / NODES 조회 실패 시).
@@ -526,17 +545,24 @@ func applyClusterConditions(
 		ObservedGeneration: vc.Generation,
 	})
 
-	// ClusterReady — cluster_state=ok && 16384 slot.
-	clusterReady := info != nil && info.State == "ok" && info.SlotsAssigned == 16384
+	// ClusterReady — cluster_state=ok && 16384 slot assigned && 모든 slot 이 ok.
+	// 결함 ⑤: slots_assigned=16384 이지만 slots_ok<16384 (fail master 소유 slot) 인
+	// partial-slot outage 를 Ready 로 오판하지 않도록 slots_ok 도 게이트에 포함.
+	clusterReady := info != nil && info.State == "ok" &&
+		info.SlotsAssigned == vk.ClusterTotalSlots && info.SlotsOK == vk.ClusterTotalSlots
 	clusterReason := "ClusterStateOK"
 	clusterMsg := ""
 	if !clusterReady {
 		clusterReason = "ClusterNotConverged"
 		if info == nil {
 			clusterMsg = "Cluster info not yet polled"
 		} else {
-			clusterMsg = fmt.Sprintf("state=%s slots_assigned=%d ready_replicas=%d/%d",
-				info.State, info.SlotsAssigned, readyReplicas, totalReplicas)
+			if info.State == "ok" && info.SlotsAssigned == vk.ClusterTotalSlots &&
+				info.SlotsOK < vk.ClusterTotalSlots {
+				clusterReason = "PartialSlotOutage"
+			}
+			clusterMsg = fmt.Sprintf("state=%s slots_assigned=%d slots_ok=%d ready_replicas=%d/%d",
+				info.State, info.SlotsAssigned, info.SlotsOK, readyReplicas, totalReplicas)
 		}
 	}
 	setCondition(conds, metav1.Condition{
@@ -633,7 +659,12 @@ func decidePhase(vc *cachev1alpha1.ValkeyCluster, readyReplicas, totalReplicas i
 		return cachev1alpha1.ClusterPhaseInitializing
 	case info == nil || info.State != "ok":
 		return cachev1alpha1.ClusterPhaseInitializing
-	case info.SlotsAssigned != 16384:
+	case info.SlotsAssigned != vk.ClusterTotalSlots:
+		return cachev1alpha1.ClusterPhaseResharding
+	case info.SlotsOK < vk.ClusterTotalSlots:
+		// 결함 ⑤ partial-slot outage: state=ok + slots_assigned=16384 이지만
+		// 일부 slot 이 ok 가 아님 (fail master 소유). Running 으로 오판하지 않고
+		// 빠른 requeue(Resharding cadence)로 self-heal(takeover)을 재시도시킨다.
 		return cachev1alpha1.ClusterPhaseResharding
 	default:
 		return cachev1alpha1.ClusterPhaseRunning

internal/valkey/replication.go

@@ -48,6 +48,22 @@ func PromoteToPrimary(ctx context.Context, c *redis.Client) error {
 	return nil
 }
 
+// ClusterFailoverTakeover — replica 에서 CLUSTER FAILOVER TAKEOVER 발행 (결함 ⑤).
+//
+// 일반 CLUSTER FAILOVER 는 master 와의 handshake (offset 동기 + 합의)를 요구하므로
+// master 가 *이미 fail* 인 partial-slot outage 에서는 동작하지 않는다. TAKEOVER 옵션은
+// 합의 없이 replica 가 즉시 slot 소유권 + epoch 를 승계해 stuck slot 을 ok 로 되돌린다.
+//
+// 주의: TAKEOVER 는 split-brain 위험이 있어 *master 가 확실히 fail* 일 때만 호출해야
+// 한다 (caller 의 보수적 게이트 책임). go-redis 의 ClusterFailover 는 옵션 인자를
+// 받지 않으므로 raw Do 로 발행한다. 본 명령은 replica 노드에 직접 보내야 한다.
+func ClusterFailoverTakeover(ctx context.Context, c *redis.Client) error {
+	if err := c.Do(ctx, "CLUSTER", "FAILOVER", "TAKEOVER").Err(); err != nil {
+		return fmt.Errorf("cluster failover takeover: %w", err)
+	}
+	return nil
+}
+
 // ParseReplicationOffset — INFO replication 응답에서 master_repl_offset 또는
 // slave_repl_offset 추출. ADR-0017 의 failover 후보 선출에 사용.
 //