alex-zaitsev · Oct 23, 2026 · Oct 22, 2026 · Oct 22, 2026 · Oct 22, 2026 · Oct 22, 2026
diff --git a/pkg/controller/chk/worker-raft-safety.go b/pkg/controller/chk/worker-raft-safety.go
@@ -416,7 +416,11 @@ func (w *worker) ensureQuorumSafeToDisruptHost(
 	if snap == nil || !snap.rolling || !ensembleHasQuorumHeadroom(snap.members) {
 		return nil
 	}
-	if !w.hostDisruptionWouldBreakQuorum(ctx, host, opts, *snap) {
+	blocked, err := w.raftPlanDisruptionBlocked(ctx, host, opts, *snap)
+	if err != nil {
+		return err
+	}
+	if !blocked {
 		return nil
 	}
 
@@ -465,7 +469,11 @@ func (w *worker) ensureQuorumSafeToDisruptHost(
 		if err := w.refreshQuorumSnapshotCounts(ctx, host, snap); err != nil {
 			return err
 		}
-		if !w.hostDisruptionWouldBreakQuorum(ctx, host, opts, *snap) {
+		blocked, err = w.raftPlanDisruptionBlocked(ctx, host, opts, *snap)
+		if err != nil {
+			return err
+		}
+		if !blocked {
 			w.a.V(1).M(host).F().Info(
 				"Raft quorum headroom available — proceeding with host %s disruption (ready=%d)",
 				host.GetName(), snap.readyCount,

diff --git a/pkg/controller/chk/worker-raftplan-enforce.go b/pkg/controller/chk/worker-raftplan-enforce.go
@@ -0,0 +1,115 @@
+// Copyright 2019 Altinity Ltd and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+package chk
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	apiChk "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse-keeper.altinity.com/v1"
+	api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1"
+	"github.com/altinity/clickhouse-operator/pkg/controller/chk/raftplan"
+	"github.com/altinity/clickhouse-operator/pkg/controller/common"
+	a "github.com/altinity/clickhouse-operator/pkg/controller/common/announcer"
+	"github.com/altinity/clickhouse-operator/pkg/controller/common/statefulset"
+)
+
+// raftPlanEnforceEnabled runs the Raft planner by default in this test
+// branch, so the standard e2e operator install needs no environment changes.
+// Setting CHOP_CHK_RAFTPLAN_ENFORCE=0 is an optional emergency rollback;
+// the existing preliminary staging, publication, waits, and rollback remain.
+func raftPlanEnforceEnabled() bool {
+	return os.Getenv("CHOP_CHK_RAFTPLAN_ENFORCE") != "0"
+}
+
+// raftPlanAdmissionVerdict is the pure-to-worker boundary. An unknown
+// membership observation cannot authorize creation of another Keeper.
+func raftPlanAdmissionVerdict(snapshot raftplan.Snapshot, id raftplan.HostID) (raftplan.AdmissionPlan, error) {
+	plan := raftplan.PlanAdmission(snapshot)
+	if plan.Reason == raftplan.ObservationUnknown {
+		return plan, fmt.Errorf("Raft admission state unknown for %s: %s", id, plan.Reason)
+	}
+	if !plan.MayCreate[id] {
+		return plan, common.ErrCRUDDeferred
+	}
+	// The existing publisher still writes membership immediately before
+	// StatefulSet creation. Ensure the planner includes the admitted host.
+	for _, included := range plan.IncludedHosts {
+		if included == id {
+			return plan, nil
+		}
+	}
+	return plan, 
8000
fmt.Errorf("Raft planner allowed creating %s without publishing its membership", id)
+}
+
+// enforceRaftPlanAdmission replaces deferUnreadyStagedAdmit for staged
+// 1->N/2->3 joins only. Existing members still reconcile while the next
+// joiner is held so they can adopt the already published Raft config.
+func (w *worker) enforceRaftPlanAdmission(ctx context.Context, host *api.Host) error {
+	if !hostIsStagedScaleUpJoin(host) || w.c == nil {
+		return nil
+	}
+	// Keep an already created STS published, including after an interrupted
+	// Create. Never re-admit it or roll its membership back.
+	own, err := w.stagedJoinerStatefulSet(ctx, host)
+	if err != nil {
+		return err
+	}
+	if own != nil {
+		return nil
+	}
+	if host.IsStopped() {
+		return errSkipStagedAdmit
+	}
+	cr, ok := host.GetCR().(*apiChk.ClickHouseKeeperInstallation)
+	if !ok {
+		return fmt.Errorf("Raft admission host %s has no CHK", host.GetName())
+	}
+	observed, err := w.observeRaftPlanAdmission(ctx, cr)
+	if err != nil {
+		return fmt.Errorf("observe Raft admission for %s: %w", host.GetName(), err)
+	}
+	id := raftPlanHostID(host)
+	state, exists := observed.Hosts[id]
+	if !exists || state.StatefulSet == raftplan.PresenceUnknown {
+		return fmt.Errorf("Raft admission observation missing StatefulSet state for %s", id)
+	}
+	if state.StatefulSet == raftplan.PresencePresent {
+		return nil // another observation saw the StatefulSet appear
+	}
+	plan, err := raftPlanAdmissionVerdict(observed, id)
+	if err == common.ErrCRUDDeferred {
+		w.a.WithEvent(host.GetCR(), a.EventActionReconcile, a.EventReasonReconcileInProgress).
+			WithAction(host.GetCR()).M(host).F().Info(
+				"Raft planner defers admitting %s: reason=%s blockedBy=%s",
+				id, plan.Reason, plan.BlockedBy,
+			)
+	}
+	return err
+}
+
+// raftPlanDisruptionBlocked supplies the existing quorum wait with its
+// authoritative predicate. A fresh Ready count is obtained by the caller
+// before each re-evaluation. Unknown or inconsistent state fails closed.
+// Without opt-in this invokes the exact legacy quorum predicate.
+func (w *worker) raftPlanDisruptionBlocked(
+	ctx context.Context, host *api.Host,
+	opts *statefulset.ReconcileOptions, snap hostEnsembleSnapshot,
+) (bool, error) {
+	if !raftPlanEnforceEnabled() {
+		return w.hostDisruptionWouldBreakQuorum(ctx, host, opts, snap), nil
+	}
+	if host == nil {
+		return true, fmt.Errorf("Raft disruption host is nil")
+	}
+	observed := w.raftPlanDisruptionSnapshot(ctx, host, opts, snap)
+	decision := raftplan.CheckDisruption(observed, raftPlanHostID(host))
+	if decision.Reason == raftplan.DisruptionStateUnknown {
+		return true, fmt.Errorf(
+			"Raft planner cannot determine disruption safety for %s (ready=%d members=%d rolling=%t)",
+			host.GetName(), snap.readyCount, snap.members, snap.rolling,
+		)
+	}
+	return !decision.Allowed, nil
+}
diff --git a/pkg/controller/chk/worker-raftplan-enforce_test.go b/pkg/controller/chk/worker-raftplan-enforce_test.go
@@ -0,0 +1,147 @@
+// Copyright 2019 Altinity Ltd and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+package chk
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	apps "k8s.io/api/apps/v1"
+
+	api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1"
+	"github.com/altinity/clickhouse-operator/pkg/apis/common/types"
+	"github.com/altinity/clickhouse-operator/pkg/controller/chk/raftplan"
+	"github.com/altinity/clickhouse-operator/pkg/controller/common"
+	a "github.com/altinity/clickhouse-operator/pkg/controller/common/announcer"
+	"github.com/altinity/clickhouse-operator/pkg/controller/common/statefulset"
+)
+
+func TestRaftPlanEnforcementDefaultForE2E(t *testing.T) {
+	// E2E deploys the operator without configuring this environment variable.
+	t.Setenv("CHOP_CHK_RAFTPLAN_ENFORCE", "")
+	require.True(t, raftPlanEnforceEnabled(), "planner must be active without e2e installation changes")
+	t.Setenv("CHOP_CHK_RAFTPLAN_ENFORCE", "1")
+	require.True(t, raftPlanEnforceEnabled())
+	// Preserve an explicit rollback for manual test deployments.
+	t.Setenv("CHOP_CHK_RAFTPLAN_ENFORCE", "0")
+	require.False(t, raftPlanEnforceEnabled())
+}
+
+func TestRaftPlanEnforcedAdmissionVerdicts(t *testing.T) {
+	ids := []raftplan.HostID{"h0", "h1", "h2"}
+	makeSnap := func() raftplan.Snapshot {
+		return raftplan.Snapshot{
+			Established: true, PublishedKnown: true,
+			DesiredHosts: ids, AncestorHosts: ids[:1], PublishedHosts: ids[:1],
+			Hosts: map[raftplan.HostID]raftplan.HostState{
+				"h0": {ID: "h0", InAncestor: true, StatefulSet: raftplan.PresencePresent, Ready: true, ReadyKnown: true},
+				"h1": {ID: "h1", StatefulSet: raftplan.PresenceAbsent, ReadyKnown: true},
+				"h2": {ID: "h2", StatefulSet: raftplan.PresenceAbsent, ReadyKnown: true},
+			},
+		}
+	}
+	t.Run("first joiner admitted, second held", func(t *testing.T) {
+		s := makeSnap()
+		plan, err := raftPlanAdmissionVerdict(s, "h1")
+		require.NoError(t, err)
+		require.True(t, plan.MayCreate["h1"])
+		_, err = raftPlanAdmissionVerdict(s, "h2")
+		require.ErrorIs(t, err, common.ErrCRUDDeferred)
+	})
+	t.Run("unready joiner blocks next", func(t *testing.T) {
+		s := makeSnap()
+		s.PublishedHosts = ids[:2]
+		s.Hosts["h1"] = raftplan.HostState{ID: "h1", StatefulSet: raftplan.PresencePresent, ReadyKnown: true}
+		plan, err := raftPlanAdmissionVerdict(s, "h2")
+		require.ErrorIs(t, err, common.ErrCRUDDeferred)
+		require.Equal(t, raftplan.HostID("h1"), plan.BlockedBy)
+	})
+	t.Run("ready joiner unblocks next", func(t *testing.T) {
+		s := makeSnap()
+		s.PublishedHosts = ids[:2]
+		s.Hosts["h1"] = raftplan.HostState{ID: "h1", StatefulSet: raftplan.PresencePresent, ReadyKnown: true, Ready: true}
+		_, err := raftPlanAdmissionVerdict(s, "h2")
+		require.NoError(t, err)
+	})
+	t.Run("unknown publication fails closed", func(t *testing.T) {
+		s := makeSnap()
+		s.PublishedKnown = false
+		_, err := raftPlanAdmissionVerdict(s, "h1")
+		require.Error(t, err)
+		require.False(t, errors.Is(err, common.ErrCRUDDeferred))
+	})
+	t.Run("unknown StatefulSet fails closed", func(t *testing.T) {
+		s := makeSnap()
+		s.Hosts["h1"] = raftplan.HostState{ID: "h1", StatefulSet: raftplan.PresenceUnknown}
+		_, err := raftPlanAdmissionVerdict(s, "h1")
+		require.Error(t, err)
+	})
+	t.Run("stopped defe
8000
rs missing member", func(t *testing.T) {
+		s := makeSnap()
+		s.Stopped = true
+		_, err := raftPlanAdmissionVerdict(s, "h1")
+		require.ErrorIs(t, err, common.ErrCRUDDeferred)
+	})
+}
+
+func TestRaftPlanEnforcedDisruptionGate(t *testing.T) {
+	// Exercise the same no-environment default used by the e2e operator install.
+	t.Setenv("CHOP_CHK_RAFTPLAN_ENFORCE", "")
+	ctx := context.Background()
+	for _, tc := range []struct {
+		name string
+		ready int
+		status types.ObjectStatus
+		forceRecreate bool
+		blocked bool
+	}{
+		{"degraded", 2, types.ObjectStatusModified, false, true},
+		{"healthy", 3, types.ObjectStatusModified, false, false},
+		{"unchanged", 2, types.ObjectStatusSame, false, false},
+		{"force recreate", 2, types.ObjectStatusSame, true, true},
+		{"new", 2, types.ObjectStatusRequested, false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			host := hostOnCR(chkWithHosts(3))
+			host.Runtime.CurStatefulSet = &apps.StatefulSet{}
+			host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1
+			host.GetReconcileAttributes().SetStatus(tc.status)
+			w := &worker{a: a.NewAnnouncer(nil, nil)}
+			snap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: tc.ready}
+			var opts *statefulset.ReconcileOptions
+			if tc.forceRecreate {
+				opts = statefulset.NewReconcileStatefulSetOptions().SetForceRecreate()
+			}
+			blocked, err := w.raftPlanDisruptionBlocked(ctx, host, opts, snap)
+			require.NoError(t, err)
+			require.Equal(t, tc.blocked, blocked)
+		})
+	}
+	t.Run("inconsistent quorum counts fail closed", func(t *testing.T) {
+		host := hostOnCR(chkWithHosts(3))
+		host.Runtime.CurStatefulSet = &apps.StatefulSet{}
+		host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1
+		host.GetReconcileAttributes().SetStatus(types.ObjectStatusModified)
+		_, err := (&worker{a: a.NewAnnouncer(nil, nil)}).raftPlanDisruptionBlocked(ctx, host, nil,
+			hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 4})
+		require.Error(t, err)
+	})
+	t.Run("existing quorum gate defers", func(t *testing.T) {
+		host := hostOnCR(chkWithHosts(3))
+		host.Runtime.CurStatefulSet = &apps.StatefulSet{}
+		host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1
+		host.GetReconcileAttributes().SetStatus(types.ObjectStatusModified)
+		w := &worker{
+			a: a.NewAnnouncer(nil, nil),
+			countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 2 },
+			quorumDisruptPollOverride: 2 * time.Millisecond,
+			quorumDisruptWaitOverride: 10 * time.Millisecond,
+		}
+		snap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 2}
+		err := w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &snap)
+		require.ErrorIs(t, err, common.ErrCRUDDeferred)
+	})
+}
diff --git a/pkg/controller/chk/worker-reconciler-chk.go b/pkg/controller/chk/worker-reconciler-chk.go
@@ -722,8 +722,15 @@ func (w *worker) reconcileHost(ctx context.Context, host *api.Host) error {
 	// for the previous joiner's Ready wait (ObjectStatusSame returns first).
 	// The next missing joiner has to be postponed here, while the loop continues
 	// so existing members can still roll.
-	admitErr := w.deferUnreadyStagedAdmit(ctx, host)
-	w.shadowRaftAdmission(ctx, host, admitErr)
+	var admitErr error
+	if raftPlanEnforceEnabled() {
+		// Opt-in: the pure planner decides serial joiner admission.
+		// Existing staging, publication, and rollback remain unchanged.
+		admitErr = w.enforceRaftPlanAdmission(ctx, host)
+	} else {
+		admitErr = w.deferUnreadyStagedAdmit(ctx, host)
+		w.shadowRaftAdmission(ctx, host, admitErr)
+	}
 	if err := admitErr; err != nil {
 		if errors.Is(err, errSkipStagedAdmit) {
 			return nil