From 2d426c45f57adf024ac4f900f41f54a8adfda50d Mon Sep 17 00:00:00 2001
From: Rain <rain@oxide.computer>
Date: Mon, 22 Jun 2026 18:03:29 -0700
Subject: [PATCH 1/4] repro for omicron#10658

---
 pkg/kv/kvserver/allocator_rf_collapse_test.go | 122 ++++++++++++++++++
 1 file changed, 122 insertions(+)
 create mode 100644 pkg/kv/kvserver/allocator_rf_collapse_test.go

diff --git a/pkg/kv/kvserver/allocator_rf_collapse_test.go b/pkg/kv/kvserver/allocator_rf_collapse_test.go
new file mode 100644
index 0000000000..04f04f8ab2
--- /dev/null
+++ b/pkg/kv/kvserver/allocator_rf_collapse_test.go
@@ -0,0 +1,122 @@
+// Copyright 2014 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package kvserver
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+// Test demonstrating the behaviors that led to omicron#10658.
+func TestAllocatorDownReplicatesOnColdLivenessCache(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+
+	var coldLiveness *liveness.NodeLiveness
+	stopper, g, mc, sp, _ := createTestStorePool(ctx,
+		TestTimeUntilStoreDeadOff, false, /* deterministic */
+		func() int {
+			if coldLiveness == nil {
+				return 0 // not reached in this test; nodeCountFn is unused before construction
+			}
+			return coldLiveness.GetNodeCount()
+		},
+		livenesspb.NodeLivenessStatus_LIVE)
+	defer stopper.Stop(ctx)
+
+	// We don't Start() the NodeLiveness, since we are about only the cache and
+	// GetNodeCount(), not the heartbeat loop.
+	clock := hlc.NewClock(mc.UnixNano, time.Nanosecond)
+	coldLiveness = liveness.NewNodeLiveness(liveness.NodeLivenessOptions{
+		AmbientCtx:              log.MakeTestingAmbientContext(stopper.Tracer()),
+		Stopper:                 stopper,
+		Settings:                cluster.MakeTestingClusterSettings(),
+		Gossip:                  g,
+		Clock:                   clock,
+		LivenessThreshold:       time.Minute,
+		RenewalDuration:         time.Second,
+		HistogramWindowInterval: time.Minute,
+	})
+
+	a := MakeAllocator(sp, func(string) (time.Duration, bool) {
+		return 0, true
+	}, nil /* knobs */, nil /* storeMetrics */)
+
+	// RF 5, with 5 healthy replicas on 5 live stores.
+	conf := roachpb.SpanConfig{NumReplicas: 5}
+	allFive := []roachpb.StoreID{1, 2, 3, 4, 5}
+
+	testCases := []struct {
+		cacheRecords        int
+		expectedNumReplicas int
+		expectedAction      AllocatorAction
+	}{
+		// < 5 leads to an effective RF of 3. (These are in ascending order to
+		// make testing easier.)
+		{cacheRecords: 0, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
+		{cacheRecords: 1, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
+		{cacheRecords: 2, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
+		{cacheRecords: 3, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
+		{cacheRecords: 4, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
+		// 5 leads to an effective RF of 5.
+		{cacheRecords: 5, expectedNumReplicas: 5, expectedAction: AllocatorConsiderRebalance},
+	}
+
+	nextNode := roachpb.NodeID(1)
+	for _, c := range testCases {
+		t.Run(fmt.Sprintf("cacheRecords=%d", c.cacheRecords), func(t *testing.T) {
+			// Grow the real cache through maybeUpdate.
+			for coldLiveness.GetNodeCount() < c.cacheRecords {
+				coldLiveness.TestingMaybeUpdate(ctx, liveness.Record{
+					Liveness: livenesspb.Liveness{
+						NodeID:     nextNode,
+						Epoch:      1,
+						Membership: livenesspb.MembershipStatus_ACTIVE,
+					},
+				})
+				nextNode++
+			}
+			require.Equalf(t, c.cacheRecords, coldLiveness.GetNodeCount(),
+				"the real liveness cache should hold exactly %d of 5 records", c.cacheRecords)
+
+			mockStorePool(sp, allFive, nil, nil, nil, nil, nil)
+			desc := makeDescriptor(allFive)
+
+			clusterNodes := a.storePool.ClusterNodeCount()
+			require.Equalf(t, c.cacheRecords, clusterNodes,
+				"ClusterNodeCount() should flow from the real NodeLiveness.GetNodeCount() (%d records)",
+				c.cacheRecords)
+
+			effectiveNumReplicas := GetNeededVoters(conf.NumReplicas, clusterNodes)
+			require.Equalf(t, c.expectedNumReplicas, effectiveNumReplicas,
+				"GetNeededVoters(5, %d) sizes a healthy RF-5 range to %d replicas",
+				clusterNodes, effectiveNumReplicas)
+
+			action, _ := a.ComputeAction(ctx, conf, &desc)
+			require.Equalf(t, c.expectedAction.String(), action.String(),
+				"5 live healthy replicas + a liveness cache of %d/5 records -> allocator action %s",
+				c.cacheRecords, action)
+		})
+	}
+}

From 18bb768bf0a36a2b674b352463b784cbd8f2689f Mon Sep 17 00:00:00 2001
From: Rain <rain@oxide.computer>
Date: Wed, 24 Jun 2026 10:42:00 -0700
Subject: [PATCH 2/4] experiment 2 (omicron#10658): policy-floor the effective
 RF
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Drive the effective replication factor exclusively by operator policy — the
configured RF and operator decommissioning — never by transient cluster health.
ComputeAction floors neededVoters at the range's own non-decommissioned voter
count (capped at the configured RF), using the authoritative range descriptor.
A dead-but-not-decommissioned voter still counts, so a cold liveness cache can no
longer size a healthy range below the replicas it has and trim it (the
omicron#10658 trigger). Lowering num_replicas or decommissioning (both operator
policy) still reduce the effective RF.

Unlike experiment 1 (dropping the downshift), this keeps GetNeededVoters intact,
so genuinely small / bringing-up clusters stay happy (no purgatory churn) and
decommissioning still completes. Verified: madrid trim becomes a no-op; a 5->3
decommission completes; small clusters and the dead-but-present control are
correct. Two DynamicNumReplicas cases shift from trimming a healthy/dead-bearing
even-4 range toward replacing/keeping replicas, per the don't-reduce-RF-for-
transient-state invariant.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 pkg/kv/kvserver/allocator.go                  |  23 +++
 pkg/kv/kvserver/allocator_policyfloor_test.go | 169 ++++++++++++++++++
 pkg/kv/kvserver/allocator_rf_collapse_test.go |  29 ++-
 pkg/kv/kvserver/allocator_test.go             |  22 ++-
 4 files changed, 227 insertions(+), 16 deletions(-)
 create mode 100644 pkg/kv/kvserver/allocator_policyfloor_test.go

diff --git a/pkg/kv/kvserver/allocator.go b/pkg/kv/kvserver/allocator.go
index 5ad1572660..2b8e9b6144 100644
--- a/pkg/kv/kvserver/allocator.go
+++ b/pkg/kv/kvserver/allocator.go
@@ -689,6 +689,29 @@ func (a *Allocator) computeAction(
 	// decommissioning/decommissioned nodes.
 	clusterNodes := a.storePool.ClusterNodeCount()
 	neededVoters := GetNeededVoters(conf.GetNumVoters(), clusterNodes)
+
+	// TODO-RAINCLAUDE: experiment 2 for omicron#10658 — policy-floor the effective
+	// RF. The effective replication factor must be driven exclusively by operator
+	// policy: the configured RF (conf.GetNumVoters()) and operator decommissioning.
+	// It must NOT be driven by transient cluster health. GetNeededVoters above
+	// derives neededVoters from clusterNodes, which is the leaseholder's cache-based
+	// node count and can read phantom-low when the liveness cache is cold (the
+	// omicron#10658 trigger). We therefore floor neededVoters at the number of this
+	// range's own voters that sit on nodes the operator has NOT decommissioned,
+	// capped at the configured RF. A dead-but-not-decommissioned voter still counts,
+	// so a cold cache can no longer size a healthy range below the replicas it
+	// already has and trim it; only lowering num_replicas or decommissioning (both
+	// operator policy) can reduce the effective RF. The range descriptor is
+	// authoritative (Raft-replicated), so this needs no KV scan.
+	if policyFloor := haveVoters - len(decommissioningVoters); policyFloor > neededVoters {
+		if maxRF := int(conf.GetNumVoters()); policyFloor > maxRF {
+			policyFloor = maxRF
+		}
+		if policyFloor > neededVoters {
+			neededVoters = policyFloor
+		}
+	}
+
 	desiredQuorum := computeQuorum(neededVoters)
 	quorum := computeQuorum(haveVoters)
 
diff --git a/pkg/kv/kvserver/allocator_policyfloor_test.go b/pkg/kv/kvserver/allocator_policyfloor_test.go
new file mode 100644
index 0000000000..b40091a422
--- /dev/null
+++ b/pkg/kv/kvserver/allocator_policyfloor_test.go
@@ -0,0 +1,169 @@
+// Copyright 2014 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package kvserver
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+// TODO-RAINCLAUDE: experiment 2 for omicron#10658. Verifies the policy-floor:
+// the effective RF is driven only by operator policy (configured RF +
+// decommissioning), never by a transient/cold-cache node count. Contrast with
+// experiment 1 (drop the downshift): the floor keeps the downshift's *good*
+// behavior — a genuinely small cluster stays happy instead of churning — while
+// still neutralizing the madrid trim.
+func TestAllocatorPolicyFloorScenarios(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	var numNodes int
+	stopper, _, _, sp, _ := createTestStorePool(ctx,
+		TestTimeUntilStoreDeadOff, false, /* deterministic */
+		func() int { return numNodes },
+		livenesspb.NodeLivenessStatus_LIVE)
+	defer stopper.Stop(ctx)
+	a := MakeAllocator(sp, func(string) (time.Duration, bool) {
+		return 0, true
+	}, nil /* knobs */, nil /* storeMetrics */)
+	conf := roachpb.SpanConfig{NumReplicas: 5}
+
+	type scenario struct {
+		name            string
+		storeList       []roachpb.StoreID
+		live            []roachpb.StoreID
+		unavailable     []roachpb.StoreID
+		dead            []roachpb.StoreID
+		decommissioning []roachpb.StoreID
+		nodeCount       int
+		wantAction      AllocatorAction
+		// what experiment 1 (bare drop of the downshift) produced, for contrast
+		exp1Action AllocatorAction
+		// what the original downshifting code produced
+		oldAction AllocatorAction
+	}
+
+	scenarios := []scenario{
+		{
+			name:       "madrid cold cache: 5 live healthy, 0 decommissioning, phantom count 3",
+			storeList:  []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:       []roachpb.StoreID{1, 2, 3, 4, 5},
+			nodeCount:  3,
+			wantAction: AllocatorConsiderRebalance, // FIXED (floored to 5)
+			exp1Action: AllocatorConsiderRebalance,
+			oldAction:  AllocatorRemoveVoter, // the bug
+		},
+		{
+			name:            "decommission: range has 5 voters, 2 decommissioning, count 3",
+			storeList:       []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:            []roachpb.StoreID{1, 2, 3},
+			decommissioning: []roachpb.StoreID{4, 5},
+			nodeCount:       3,
+			wantAction:      AllocatorRemoveDecommissioningVoter, // proceeds
+			exp1Action:      AllocatorReplaceDecommissioningVoter,
+			oldAction:       AllocatorRemoveDecommissioningVoter,
+		},
+		{
+			name:       "small cluster: RF5 range on 3 nodes, only 3 exist, 0 decommissioning",
+			storeList:  []roachpb.StoreID{1, 2, 3},
+			live:       []roachpb.StoreID{1, 2, 3},
+			nodeCount:  3,
+			wantAction: AllocatorConsiderRebalance, // happy — NO churn (floor==downshift==3)
+			exp1Action: AllocatorAddVoter,          // experiment 1 churned here
+			oldAction:  AllocatorConsiderRebalance,
+		},
+		{
+			name:       "dead-but-present (control): 2 dead, warm cache counts 5",
+			storeList:  []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:       []roachpb.StoreID{1, 2, 3},
+			dead:       []roachpb.StoreID{4, 5},
+			nodeCount:  5,
+			wantAction: AllocatorReplaceDeadVoter,
+			exp1Action: AllocatorReplaceDeadVoter,
+			oldAction:  AllocatorReplaceDeadVoter,
+		},
+		{
+			name:       "dead + cold cache: 2 dead (NOT decommissioned), phantom count 3",
+			storeList:  []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:       []roachpb.StoreID{1, 2, 3},
+			dead:       []roachpb.StoreID{4, 5},
+			nodeCount:  3,
+			wantAction: AllocatorReplaceDeadVoter, // floored to 5: keep trying to maintain RF, never trim
+			exp1Action: AllocatorReplaceDeadVoter,
+			oldAction:  AllocatorRemoveDeadVoter, // old code trimmed toward 3
+		},
+	}
+
+	for _, s := range scenarios {
+		t.Run(s.name, func(t *testing.T) {
+			numNodes = s.nodeCount
+			mockStorePool(sp, s.live, s.unavailable, s.dead, s.decommissioning, nil, nil)
+			desc := makeDescriptor(s.storeList)
+			desc.EndKey = roachpb.RKey(keys.SystemPrefix)
+			action, _ := a.ComputeAction(ctx, conf, &desc)
+			require.Equalf(t, s.wantAction.String(), action.String(),
+				"policy-floor wants %s (experiment 1 gave %s; original downshift gave %s)",
+				s.wantAction, s.exp1Action, s.oldAction)
+		})
+	}
+}
+
+// TODO-RAINCLAUDE: experiment 2 — a full decommission of two nodes from a 5-node
+// RF-5 cluster must still complete under the policy floor. Each step removes a
+// decommissioning voter until the range settles at 3 and is happy.
+func TestAllocatorPolicyFloorDecommissionCompletes(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	var numNodes int
+	stopper, _, _, sp, _ := createTestStorePool(ctx,
+		TestTimeUntilStoreDeadOff, false, /* deterministic */
+		func() int { return numNodes },
+		livenesspb.NodeLivenessStatus_LIVE)
+	defer stopper.Stop(ctx)
+	a := MakeAllocator(sp, func(string) (time.Duration, bool) {
+		return 0, true
+	}, nil /* knobs */, nil /* storeMetrics */)
+	conf := roachpb.SpanConfig{NumReplicas: 5}
+
+	// Operator decommissions nodes 4 and 5 of a 5-node cluster. 3 nodes remain
+	// (count excludes decommissioning), so the cluster genuinely shrinks below RF.
+	numNodes = 3
+
+	steps := []struct {
+		storeList       []roachpb.StoreID
+		decommissioning []roachpb.StoreID
+		live            []roachpb.StoreID
+		wantAction      AllocatorAction
+	}{
+		{[]roachpb.StoreID{1, 2, 3, 4, 5}, []roachpb.StoreID{4, 5}, []roachpb.StoreID{1, 2, 3}, AllocatorRemoveDecommissioningVoter},
+		{[]roachpb.StoreID{1, 2, 3, 5}, []roachpb.StoreID{5}, []roachpb.StoreID{1, 2, 3}, AllocatorRemoveDecommissioningVoter},
+		{[]roachpb.StoreID{1, 2, 3}, nil, []roachpb.StoreID{1, 2, 3}, AllocatorConsiderRebalance}, // settled
+	}
+
+	for i, s := range steps {
+		mockStorePool(sp, s.live, nil, nil, s.decommissioning, nil, nil)
+		desc := makeDescriptor(s.storeList)
+		desc.EndKey = roachpb.RKey(keys.SystemPrefix)
+		action, _ := a.ComputeAction(ctx, conf, &desc)
+		require.Equalf(t, s.wantAction.String(), action.String(), "decommission step %d", i)
+	}
+}
diff --git a/pkg/kv/kvserver/allocator_rf_collapse_test.go b/pkg/kv/kvserver/allocator_rf_collapse_test.go
index 04f04f8ab2..69770e8051 100644
--- a/pkg/kv/kvserver/allocator_rf_collapse_test.go
+++ b/pkg/kv/kvserver/allocator_rf_collapse_test.go
@@ -67,20 +67,31 @@ func TestAllocatorDownReplicatesOnColdLivenessCache(t *testing.T) {
 	conf := roachpb.SpanConfig{NumReplicas: 5}
 	allFive := []roachpb.StoreID{1, 2, 3, 4, 5}
 
+	// TODO-RAINCLAUDE: experiment 2 for omicron#10658 (policy-floor the effective
+	// RF). Note the difference from experiment 1 (which removed the downshift from
+	// GetNeededVoters): here GetNeededVoters is UNCHANGED, so `expectedNumReplicas`
+	// still reflects the downshift — a cold cache still computes an effective RF of
+	// 3. What changed is ComputeAction: it floors the target at the range's own
+	// non-decommissioned replica count (5 here), so a healthy RF-5 range is no
+	// longer trimmed even though GetNeededVoters reports 3. The bug was the
+	// *removal*, and the floor blocks it while leaving the downshift formula (and
+	// thus the small-cluster and decommission behavior) intact. `oldExpectedAction`
+	// is the original buggy action.
 	testCases := []struct {
 		cacheRecords        int
 		expectedNumReplicas int
+		oldExpectedAction   AllocatorAction
 		expectedAction      AllocatorAction
 	}{
-		// < 5 leads to an effective RF of 3. (These are in ascending order to
-		// make testing easier.)
-		{cacheRecords: 0, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
-		{cacheRecords: 1, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
-		{cacheRecords: 2, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
-		{cacheRecords: 3, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
-		{cacheRecords: 4, expectedNumReplicas: 3, expectedAction: AllocatorRemoveVoter},
-		// 5 leads to an effective RF of 5.
-		{cacheRecords: 5, expectedNumReplicas: 5, expectedAction: AllocatorConsiderRebalance},
+		// GetNeededVoters still downshifts to 3 on a cold cache, but ComputeAction
+		// floors to the 5 live non-decommissioned replicas and does nothing.
+		{cacheRecords: 0, expectedNumReplicas: 3, oldExpectedAction: AllocatorRemoveVoter, expectedAction: AllocatorConsiderRebalance},
+		{cacheRecords: 1, expectedNumReplicas: 3, oldExpectedAction: AllocatorRemoveVoter, expectedAction: AllocatorConsiderRebalance},
+		{cacheRecords: 2, expectedNumReplicas: 3, oldExpectedAction: AllocatorRemoveVoter, expectedAction: AllocatorConsiderRebalance},
+		{cacheRecords: 3, expectedNumReplicas: 3, oldExpectedAction: AllocatorRemoveVoter, expectedAction: AllocatorConsiderRebalance},
+		{cacheRecords: 4, expectedNumReplicas: 3, oldExpectedAction: AllocatorRemoveVoter, expectedAction: AllocatorConsiderRebalance},
+		// A warm cache (all 5 records) was correct before and after.
+		{cacheRecords: 5, expectedNumReplicas: 5, oldExpectedAction: AllocatorConsiderRebalance, expectedAction: AllocatorConsiderRebalance},
 	}
 
 	nextNode := roachpb.NodeID(1)
diff --git a/pkg/kv/kvserver/allocator_test.go b/pkg/kv/kvserver/allocator_test.go
index add12a9cb0..111f16789d 100644
--- a/pkg/kv/kvserver/allocator_test.go
+++ b/pkg/kv/kvserver/allocator_test.go
@@ -6683,13 +6683,15 @@ func TestAllocatorComputeActionDynamicNumReplicas(t *testing.T) {
 			decommissioning:     []roachpb.StoreID{1, 2, 3},
 		},
 		{
-			// Four live stores and one dead one, so the effective replication
-			// factor would be even (four), in which case we drop down one more
-			// to three. Then the right thing becomes removing the dead replica
-			// from the range at hand, rather than trying to replace it.
+			// Four replicas, one dead, none decommissioning. GetNeededVoters still
+			// downshifts to 3 (expectedNumReplicas), but the policy floor (experiment
+			// 2) holds the effective target at the 4 non-decommissioned replicas, so
+			// ComputeAction *replaces* the dead voter rather than trimming to 3.
+			// Deadness is transient and is not operator policy, so the RF is not
+			// reduced for it. (Was: AllocatorRemoveDeadVoter.)
 			storeList:           []roachpb.StoreID{1, 2, 3, 4},
 			expectedNumReplicas: 3,
-			expectedAction:      AllocatorRemoveDeadVoter,
+			expectedAction:      AllocatorReplaceDeadVoter,
 			live:                []roachpb.StoreID{1, 2, 3, 5},
 			unavailable:         nil,
 			dead:                []roachpb.StoreID{4},
@@ -6741,10 +6743,16 @@ func TestAllocatorComputeActionDynamicNumReplicas(t *testing.T) {
 			decommissioning:     nil,
 		},
 		{
-			// Three again, on account of avoiding the even four.
+			// Four healthy replicas, none decommissioning. The original code
+			// trimmed the even four down to three (GetNeededVoters still reports 3,
+			// expectedNumReplicas). The policy floor (experiment 2) holds the target
+			// at the 4 non-decommissioned replicas, so ComputeAction leaves the
+			// range alone: it will not remove a healthy replica to satisfy a
+			// transiently-derived target. The even-quorum nicety yields to the
+			// don't-trim-healthy-replicas invariant. (Was: AllocatorRemoveVoter.)
 			storeList:           []roachpb.StoreID{1, 2, 3, 4},
 			expectedNumReplicas: 3,
-			expectedAction:      AllocatorRemoveVoter,
+			expectedAction:      AllocatorConsiderRebalance,
 			live:                []roachpb.StoreID{1, 2, 3, 4},
 			unavailable:         nil,
 			dead:                nil,

From 1ef42e7c7792d955c8585748c449eda42bdedc7c Mon Sep 17 00:00:00 2001
From: Rain <rain@oxide.computer>
Date: Mon, 6 Jul 2026 12:02:10 -0700
Subject: [PATCH 3/4] fix policy floor (omicron#10658): membership-keyed
 exclusion; floor the gauges

Two fixes from the policy-floor review (policy-floor-review.md in the
findings repo):

1. Key the floor exclusion on liveness membership, not store-pool status.
   The floor subtracted decommissioningVoters, which keys on
   storeStatusDecommissioning and therefore only matches LIVE decommissioning
   nodes. A dead node under decommission reports DECOMMISSIONED (dead +
   non-active membership), which classifies as storeStatusDead, so it kept
   counting toward the floor: its replica could only ever be replaced, never
   shed, and on a cluster with no spare node the replacement has no
   allocation target, stalling the decommission in purgatory forever. This
   violated the floor's own principle: decommissioning is operator policy
   and must be able to lower the effective RF regardless of node health.
   Subtract replicas whose NodeLivenessStatus is DECOMMISSIONING or
   DECOMMISSIONED instead: membership is the policy bit, store status mixes
   health back in. UNAVAILABLE decommissioning nodes stay unsubtracted
   deliberately, since excluding them while they sit in neither the dead nor
   the decommissioning status set would open a liveness-blind RemoveVoter
   window; they resolve to live or dead within the store-dead threshold. The
   invariant is preserved: RemoveVoter still fires only when haveVoters
   exceeds the configured RF.

2. Apply the same floor in calcRangeCounter so the gauges agree with the
   allocator. Without it, a phantom-low count blinds ranges_underreplicated
   exactly when it matters, false-alarms ranges_overreplicated on every
   healthy RF-5 range, and reports states the allocator deliberately
   preserves (4 voters on 4 nodes after a shrink) as over-replicated
   forever. With it, a healthy range whose voters the leaseholder cannot
   account for reads under-replicated: the honest signal that the view is
   degraded. Genuine over-replication versus the configured RF is still
   detected via the cap.

Verified: TestAllocator*, TestStorePool*, *Decommission*,
TestReplicateQueue*, *Metrics* all pass. New coverage: dead-node
decommission with and without a spare node (completes in both; settles at 4
voters without a spare, at the configured RF 5 with one), the
unavailable-decommissioning pause, the even steady state, and the floored
gauges.
---
 pkg/kv/kvserver/allocator.go                  |  26 ++-
 pkg/kv/kvserver/allocator_policyfloor_test.go | 159 +++++++++++++++++-
 pkg/kv/kvserver/replica_metrics.go            |  28 +++
 pkg/kv/kvserver/replica_metrics_test.go       | 123 ++++++++++++++
 pkg/kv/kvserver/store_pool.go                 |  29 ++++
 5 files changed, 354 insertions(+), 11 deletions(-)

diff --git a/pkg/kv/kvserver/allocator.go b/pkg/kv/kvserver/allocator.go
index 2b8e9b6144..ebad54a1d8 100644
--- a/pkg/kv/kvserver/allocator.go
+++ b/pkg/kv/kvserver/allocator.go
@@ -697,13 +697,25 @@ func (a *Allocator) computeAction(
 	// derives neededVoters from clusterNodes, which is the leaseholder's cache-based
 	// node count and can read phantom-low when the liveness cache is cold (the
 	// omicron#10658 trigger). We therefore floor neededVoters at the number of this
-	// range's own voters that sit on nodes the operator has NOT decommissioned,
-	// capped at the configured RF. A dead-but-not-decommissioned voter still counts,
-	// so a cold cache can no longer size a healthy range below the replicas it
-	// already has and trim it; only lowering num_replicas or decommissioning (both
-	// operator policy) can reduce the effective RF. The range descriptor is
-	// authoritative (Raft-replicated), so this needs no KV scan.
-	if policyFloor := haveVoters - len(decommissioningVoters); policyFloor > neededVoters {
+	// range's own voters that sit on nodes the operator has NOT directed out of the
+	// cluster, capped at the configured RF. The excluded set is keyed on liveness
+	// MEMBERSHIP (DECOMMISSIONING or DECOMMISSIONED), not on the decommissioningVoters
+	// status set above: a dead node under decommission reports DECOMMISSIONED, which
+	// classifies as storeStatusDead, and it must still lower the floor — otherwise
+	// its replica could only ever be replaced, never shed, and on a cluster with no
+	// spare node the decommission would stall in purgatory forever. A
+	// dead-but-not-decommissioned voter still counts toward the floor, so a cold
+	// cache can never size a healthy range below the replicas it already has and
+	// trim it; only lowering num_replicas or decommissioning (both operator policy)
+	// can reduce the effective RF. Safety of the subtraction: whenever it drops the
+	// floor below haveVoters, the excluded replicas are by construction either
+	// live-decommissioning (handled by the RemoveDecommissioningVoter branch below)
+	// or dead (handled by RemoveDeadVoter, whose removal candidates the replicate
+	// queue restricts to dead replicas), so the liveness-blind RemoveVoter branch
+	// still fires only when haveVoters exceeds the configured RF. The range
+	// descriptor is authoritative (Raft-replicated), so this needs no KV scan.
+	policyRemovedVoters := a.storePool.decommissioningOrDecommissionedReplicas(voterReplicas)
+	if policyFloor := haveVoters - len(policyRemovedVoters); policyFloor > neededVoters {
 		if maxRF := int(conf.GetNumVoters()); policyFloor > maxRF {
 			policyFloor = maxRF
 		}
diff --git a/pkg/kv/kvserver/allocator_policyfloor_test.go b/pkg/kv/kvserver/allocator_policyfloor_test.go
index b40091a422..0037f356fb 100644
--- a/pkg/kv/kvserver/allocator_policyfloor_test.go
+++ b/pkg/kv/kvserver/allocator_policyfloor_test.go
@@ -28,7 +28,10 @@ import (
 // decommissioning), never by a transient/cold-cache node count. Contrast with
 // experiment 1 (drop the downshift): the floor keeps the downshift's *good*
 // behavior — a genuinely small cluster stays happy instead of churning — while
-// still neutralizing the madrid trim.
+// still neutralizing the madrid trim. The floor is keyed on liveness membership
+// (DECOMMISSIONING or DECOMMISSIONED), not store-pool status, so that a DEAD
+// node under decommission (which reports DECOMMISSIONED -> storeStatusDead)
+// still lowers it; see the dead-node decommission scenarios.
 func TestAllocatorPolicyFloorScenarios(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -52,8 +55,12 @@ func TestAllocatorPolicyFloorScenarios(t *testing.T) {
 		unavailable     []roachpb.StoreID
 		dead            []roachpb.StoreID
 		decommissioning []roachpb.StoreID
-		nodeCount       int
-		wantAction      AllocatorAction
+		// decommissioned mocks NodeLivenessStatus_DECOMMISSIONED, which is what a
+		// DEAD node with DECOMMISSIONING (or DECOMMISSIONED) membership reports —
+		// it classifies as storeStatusDead, not storeStatusDecommissioning.
+		decommissioned []roachpb.StoreID
+		nodeCount      int
+		wantAction     AllocatorAction
 		// what experiment 1 (bare drop of the downshift) produced, for contrast
 		exp1Action AllocatorAction
 		// what the original downshifting code produced
@@ -109,12 +116,75 @@ func TestAllocatorPolicyFloorScenarios(t *testing.T) {
 			exp1Action: AllocatorReplaceDeadVoter,
 			oldAction:  AllocatorRemoveDeadVoter, // old code trimmed toward 3
 		},
+		{
+			// A node died permanently and the operator decommissions it. The
+			// membership-keyed floor excludes it (floor 4, not 5), so the dead
+			// replica is REMOVED and the decommission completes even with no spare
+			// node. The status-keyed floor (the original experiment 2 formula) kept
+			// needed at 5 and produced ReplaceDeadVoter, which has no allocation
+			// target on the 4-node remainder: purgatory forever, and the
+			// decommission never reaches zero replicas.
+			name:           "dead-node decommission, no spare: 1 decommissioned(dead), count 4",
+			storeList:      []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:           []roachpb.StoreID{1, 2, 3, 4},
+			decommissioned: []roachpb.StoreID{5},
+			nodeCount:      4,
+			wantAction:     AllocatorRemoveDeadVoter,
+			exp1Action:     AllocatorReplaceDeadVoter, // stalls: no 5th node
+			oldAction:      AllocatorRemoveDeadVoter,
+		},
+		{
+			// Same, but a spare node 6 exists (the Oxide add-then-decommission
+			// flow). The warm count keeps needed at 5 above the floor of 4, so the
+			// dead replica is replaced onto the spare and the range stays at the
+			// configured RF.
+			name:           "dead-node decommission, spare exists: count 5",
+			storeList:      []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:           []roachpb.StoreID{1, 2, 3, 4, 6},
+			decommissioned: []roachpb.StoreID{5},
+			nodeCount:      5,
+			wantAction:     AllocatorReplaceDeadVoter,
+			exp1Action:     AllocatorReplaceDeadVoter,
+			oldAction:      AllocatorReplaceDeadVoter,
+		},
+		{
+			// A decommissioning node that is transiently unreachable reports
+			// UNAVAILABLE (LivenessStatus collapses membership for expired-but-not-
+			// yet-dead nodes) -> storeStatusUnknown. It is deliberately NOT excluded
+			// from the floor: the decommission pauses until the node resolves to
+			// live (remove decommissioning) or dead (remove dead). Excluding it
+			// would open a liveness-blind RemoveVoter window. Note the old code
+			// trimmed a LIVE voter here — the madrid mechanism without any cache
+			// coldness on the affected node.
+			name:        "decommissioning node transiently unavailable, phantom count 4",
+			storeList:   []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:        []roachpb.StoreID{1, 2, 3, 4},
+			unavailable: []roachpb.StoreID{5},
+			nodeCount:   4,
+			wantAction:  AllocatorConsiderRebalance, // pause, fail-safe
+			exp1Action:  AllocatorConsiderRebalance,
+			oldAction:   AllocatorRemoveVoter, // blind trim of a live voter
+		},
+		{
+			// The even steady state after a 5->4 shrink: the floor holds 4 healthy
+			// voters on a 4-node cluster rather than trimming to 3. The even-quorum
+			// nicety yields to the don't-trim-healthy-replicas invariant; the
+			// operator escape hatch is lowering num_replicas (policy, honored via
+			// the cap).
+			name:       "even steady state: 4 healthy voters, 4 nodes",
+			storeList:  []roachpb.StoreID{1, 2, 3, 4},
+			live:       []roachpb.StoreID{1, 2, 3, 4},
+			nodeCount:  4,
+			wantAction: AllocatorConsiderRebalance,
+			exp1Action: AllocatorAddVoter,    // wants a 5th node that does not exist
+			oldAction:  AllocatorRemoveVoter, // trimmed the even 4 to 3
+		},
 	}
 
 	for _, s := range scenarios {
 		t.Run(s.name, func(t *testing.T) {
 			numNodes = s.nodeCount
-			mockStorePool(sp, s.live, s.unavailable, s.dead, s.decommissioning, nil, nil)
+			mockStorePool(sp, s.live, s.unavailable, s.dead, s.decommissioning, s.decommissioned, nil)
 			desc := makeDescriptor(s.storeList)
 			desc.EndKey = roachpb.RKey(keys.SystemPrefix)
 			action, _ := a.ComputeAction(ctx, conf, &desc)
@@ -167,3 +237,84 @@ func TestAllocatorPolicyFloorDecommissionCompletes(t *testing.T) {
 		require.Equalf(t, s.wantAction.String(), action.String(), "decommission step %d", i)
 	}
 }
+
+// TODO-RAINCLAUDE: membership-keyed floor fix — decommissioning a DEAD node
+// must complete. Without a spare node the dead replica is removed and the range
+// settles at 4 voters (the even state the floor preserves); with a spare it is
+// replaced and the range stays at the configured RF 5. Before the fix the
+// no-spare path returned ReplaceDeadVoter forever (no allocation target ->
+// purgatory) and the node could never finish decommissioning.
+func TestAllocatorPolicyFloorDeadNodeDecommission(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	var numNodes int
+	stopper, _, _, sp, _ := createTestStorePool(ctx,
+		TestTimeUntilStoreDeadOff, false, /* deterministic */
+		func() int { return numNodes },
+		livenesspb.NodeLivenessStatus_LIVE)
+	defer stopper.Stop(ctx)
+	a := MakeAllocator(sp, func(string) (time.Duration, bool) {
+		return 0, true
+	}, nil /* knobs */, nil /* storeMetrics */)
+	conf := roachpb.SpanConfig{NumReplicas: 5}
+
+	steps := []struct {
+		name           string
+		storeList      []roachpb.StoreID
+		live           []roachpb.StoreID
+		decommissioned []roachpb.StoreID
+		nodeCount      int
+		wantAction     AllocatorAction
+	}{
+		// No spare: node 5 is dead and being decommissioned on what is now a
+		// 4-node cluster. The dead replica is removed, then the range settles at
+		// the even 4-voter state.
+		{
+			name:           "no spare: shed the dead replica",
+			storeList:      []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:           []roachpb.StoreID{1, 2, 3, 4},
+			decommissioned: []roachpb.StoreID{5},
+			nodeCount:      4,
+			wantAction:     AllocatorRemoveDeadVoter,
+		},
+		{
+			name:           "no spare: settled at 4",
+			storeList:      []roachpb.StoreID{1, 2, 3, 4},
+			live:           []roachpb.StoreID{1, 2, 3, 4},
+			decommissioned: []roachpb.StoreID{5},
+			nodeCount:      4,
+			wantAction:     AllocatorConsiderRebalance,
+		},
+		// With a spare (the Oxide add-then-decommission flow): the dead replica is
+		// replaced onto node 6, then the range is settled at the configured RF.
+		{
+			name:           "spare: replace the dead replica",
+			storeList:      []roachpb.StoreID{1, 2, 3, 4, 5},
+			live:           []roachpb.StoreID{1, 2, 3, 4, 6},
+			decommissioned: []roachpb.StoreID{5},
+			nodeCount:      5,
+			wantAction:     AllocatorReplaceDeadVoter,
+		},
+		{
+			name:           "spare: settled at RF 5",
+			storeList:      []roachpb.StoreID{1, 2, 3, 4, 6},
+			live:           []roachpb.StoreID{1, 2, 3, 4, 6},
+			decommissioned: []roachpb.StoreID{5},
+			nodeCount:      5,
+			wantAction:     AllocatorConsiderRebalance,
+		},
+	}
+
+	for _, s := range steps {
+		t.Run(s.name, func(t *testing.T) {
+			numNodes = s.nodeCount
+			mockStorePool(sp, s.live, nil, nil, nil, s.decommissioned, nil)
+			desc := makeDescriptor(s.storeList)
+			desc.EndKey = roachpb.RKey(keys.SystemPrefix)
+			action, _ := a.ComputeAction(ctx, conf, &desc)
+			require.Equalf(t, s.wantAction.String(), action.String(), "step %q", s.name)
+		})
+	}
+}
diff --git a/pkg/kv/kvserver/replica_metrics.go b/pkg/kv/kvserver/replica_metrics.go
index 7f0e8e6353..1213abd76a 100644
--- a/pkg/kv/kvserver/replica_metrics.go
+++ b/pkg/kv/kvserver/replica_metrics.go
@@ -180,6 +180,34 @@ func calcRangeCounter(
 	// unavailable ranges for each range based on the liveness table.
 	if rangeCounter {
 		neededVoters := GetNeededVoters(numVoters, clusterNodes)
+		// TODO-RAINCLAUDE: omicron#10658 — the same policy floor computeAction
+		// applies, so the gauges agree with the allocator. Without it, a phantom-low
+		// clusterNodes makes ranges_underreplicated go blind exactly when it matters
+		// and false-alarms ranges_overreplicated on every healthy RF-5 range; a
+		// cluster legitimately settled above the downshifted target (e.g. 4 voters
+		// on 4 nodes after a decommission) reads over-replicated forever. The floor
+		// counts this range's voters on membership-active nodes, capped at the
+		// configured RF. A voter absent from the liveness map counts toward the
+		// floor (fail-safe: an unverifiable replica must not lower the target), so
+		// during a cold-cache window a healthy range reads as under-replicated —
+		// the honest signal that the leaseholder cannot account for its replicas —
+		// rather than over-replicated. Metric-only: keyed on raw membership, which
+		// may diverge from the allocator's floor for the brief window where a
+		// decommissioning node is unavailable but not yet dead.
+		voterFloor := 0
+		for _, rd := range desc.Replicas().VoterDescriptors() {
+			if entry, ok := livenessMap[rd.NodeID]; !ok || entry.Membership.Active() {
+				voterFloor++
+			}
+		}
+		if voterFloor > neededVoters {
+			if maxRF := int(numVoters); voterFloor > maxRF {
+				voterFloor = maxRF
+			}
+			if voterFloor > neededVoters {
+				neededVoters = voterFloor
+			}
+		}
 		neededNonVoters := GetNeededNonVoters(int(numVoters), int(numReplicas-numVoters), clusterNodes)
 		status := desc.Replicas().ReplicationStatus(func(rDesc roachpb.ReplicaDescriptor) bool {
 			return livenessMap[rDesc.NodeID].IsLive
diff --git a/pkg/kv/kvserver/replica_metrics_test.go b/pkg/kv/kvserver/replica_metrics_test.go
index ff2b840217..f8ecd18271 100644
--- a/pkg/kv/kvserver/replica_metrics_test.go
+++ b/pkg/kv/kvserver/replica_metrics_test.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -138,6 +139,128 @@ func TestCalcRangeCounterIsLiveMap(t *testing.T) {
 	}
 }
 
+// TODO-RAINCLAUDE: omicron#10658 — the policy floor applied to the gauges, so
+// they agree with the allocator. Covers: the madrid phantom window (healthy
+// range no longer reads over-replicated; a range whose voters the leaseholder
+// cannot account for reads under-replicated instead of quiet), the
+// dead-decommission and even-steady states (no longer permanently
+// over-replicated), and genuine over-replication versus the configured RF
+// (still detected — the floor is capped at numVoters).
+func TestCalcRangeCounterPolicyFloor(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	leaseStatus := kvserverpb.LeaseStatus{
+		Lease: roachpb.Lease{
+			Replica: roachpb.ReplicaDescriptor{
+				NodeID:  1,
+				StoreID: 10,
+			},
+		},
+		State: kvserverpb.LeaseState_VALID,
+	}
+
+	fiveVoters := roachpb.NewRangeDescriptor(123, roachpb.RKeyMin, roachpb.RKeyMax,
+		roachpb.MakeReplicaSet([]roachpb.ReplicaDescriptor{
+			{NodeID: 1, StoreID: 10, ReplicaID: 1, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 2, StoreID: 20, ReplicaID: 2, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 3, StoreID: 30, ReplicaID: 3, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 4, StoreID: 40, ReplicaID: 4, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 5, StoreID: 50, ReplicaID: 5, Type: roachpb.ReplicaTypeVoterFull()},
+		}))
+
+	fourVoters := roachpb.NewRangeDescriptor(124, roachpb.RKeyMin, roachpb.RKeyMax,
+		roachpb.MakeReplicaSet([]roachpb.ReplicaDescriptor{
+			{NodeID: 1, StoreID: 10, ReplicaID: 1, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 2, StoreID: 20, ReplicaID: 2, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 3, StoreID: 30, ReplicaID: 3, Type: roachpb.ReplicaTypeVoterFull()},
+			{NodeID: 4, StoreID: 40, ReplicaID: 4, Type: roachpb.ReplicaTypeVoterFull()},
+		}))
+
+	live := liveness.IsLiveMapEntry{IsLive: true}
+	// A dead node under operator decommission: non-active membership, not live.
+	deadDecommissioning := liveness.IsLiveMapEntry{
+		Liveness: livenesspb.Liveness{Membership: livenesspb.MembershipStatus_DECOMMISSIONING},
+		IsLive:   false,
+	}
+
+	{
+		// The madrid phantom window: 5 healthy voters, all live and
+		// membership-active, but clusterNodes reads phantom-low. The floor holds
+		// needed at 5, so the range is neither over- nor under-replicated.
+		// Without the floor this read over-replicated (needed 3 < live 5).
+		ctr, down, under, over := calcRangeCounter(10, fiveVoters, leaseStatus, liveness.IsLiveMap{
+			1: live, 2: live, 3: live, 4: live, 5: live,
+		}, 5 /* numVoters */, 5 /* numReplicas */, 3 /* clusterNodes */)
+
+		require.True(t, ctr)
+		require.False(t, down)
+		require.False(t, under)
+		require.False(t, over)
+	}
+
+	{
+		// Phantom window with two voters absent from the liveness map (the actual
+		// madrid state on n1). Absent voters still count toward the floor, so
+		// needed stays 5 against 3 live: under-replicated — the honest signal that
+		// the leaseholder cannot account for two of its replicas. Without the
+		// floor this read quiet (needed 3 == live 3): blind exactly when it
+		// mattered.
+		ctr, down, under, over := calcRangeCounter(10, fiveVoters, leaseStatus, liveness.IsLiveMap{
+			1: live, 2: live, 3: live,
+		}, 5 /* numVoters */, 5 /* numReplicas */, 3 /* clusterNodes */)
+
+		require.True(t, ctr)
+		require.False(t, down)
+		require.True(t, under)
+		require.False(t, over)
+	}
+
+	{
+		// Dead-node decommission in progress: the non-active membership excludes
+		// node 5 from the floor (4, matching the allocator's target), so the range
+		// reads neither over- nor under-replicated while the dead replica is shed.
+		// Without the floor this read over-replicated (needed 3 < live 4) while
+		// the allocator was still repairing.
+		ctr, down, under, over := calcRangeCounter(10, fiveVoters, leaseStatus, liveness.IsLiveMap{
+			1: live, 2: live, 3: live, 4: live, 5: deadDecommissioning,
+		}, 5 /* numVoters */, 5 /* numReplicas */, 4 /* clusterNodes */)
+
+		require.True(t, ctr)
+		require.False(t, down)
+		require.False(t, under)
+		require.False(t, over)
+	}
+
+	{
+		// The even steady state: 4 healthy voters on a 4-node cluster. The floor
+		// holds needed at 4, so the state the allocator deliberately preserves is
+		// not reported as permanently over-replicated (needed 3 < live 4 before).
+		ctr, down, under, over := calcRangeCounter(10, fourVoters, leaseStatus, liveness.IsLiveMap{
+			1: live, 2: live, 3: live, 4: live,
+		}, 5 /* numVoters */, 5 /* numReplicas */, 4 /* clusterNodes */)
+
+		require.True(t, ctr)
+		require.False(t, down)
+		require.False(t, under)
+		require.False(t, over)
+	}
+
+	{
+		// Genuine over-replication versus the configured RF is still detected:
+		// the floor is capped at numVoters, so 5 live voters against a configured
+		// RF of 3 reads over-replicated.
+		ctr, down, under, over := calcRangeCounter(10, fiveVoters, leaseStatus, liveness.IsLiveMap{
+			1: live, 2: live, 3: live, 4: live, 5: live,
+		}, 3 /* numVoters */, 3 /* numReplicas */, 5 /* clusterNodes */)
+
+		require.True(t, ctr)
+		require.False(t, down)
+		require.False(t, under)
+		require.True(t, over)
+	}
+}
+
 func TestCalcRangeCounterLeaseHolder(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
diff --git a/pkg/kv/kvserver/store_pool.go b/pkg/kv/kvserver/store_pool.go
index 751f59264a..1e5d6656b7 100644
--- a/pkg/kv/kvserver/store_pool.go
+++ b/pkg/kv/kvserver/store_pool.go
@@ -598,6 +598,35 @@ func (sp *StorePool) decommissioningReplicas(
 	return
 }
 
+// TODO-RAINCLAUDE: fix for the omicron#10658 policy floor — the set of replicas
+// the operator has directed out of the cluster, keyed on liveness MEMBERSHIP
+// rather than store-pool status. decommissioningReplicas above only matches
+// storeStatusDecommissioning, which requires a LIVE decommissioning node: a
+// dead node under decommission reports NodeLivenessStatus_DECOMMISSIONED
+// (dead + non-active membership) and lands in storeStatusDead instead. The
+// policy floor must exclude both, or a dead node's replicas can never be shed.
+// An UNAVAILABLE node (expired but not yet past the dead threshold) is
+// deliberately NOT matched even if its membership is non-active: excluding it
+// from the floor while it sits in neither the dead nor the decommissioning
+// status set would open a liveness-blind RemoveVoter window in computeAction.
+// A node absent from the liveness cache reads as UNKNOWN and is likewise not
+// matched, which keeps the floor high — the fail-safe direction.
+func (sp *StorePool) decommissioningOrDecommissionedReplicas(
+	repls []roachpb.ReplicaDescriptor,
+) (inactive []roachpb.ReplicaDescriptor) {
+	now := sp.clock.Now().GoTime()
+	timeUntilStoreDead := TimeUntilStoreDead.Get(&sp.st.SV)
+
+	for _, repl := range repls {
+		switch sp.nodeLivenessFn(repl.NodeID, now, timeUntilStoreDead) {
+		case livenesspb.NodeLivenessStatus_DECOMMISSIONING,
+			livenesspb.NodeLivenessStatus_DECOMMISSIONED:
+			inactive = append(inactive, repl)
+		}
+	}
+	return inactive
+}
+
 // ClusterNodeCount returns the number of nodes that are possible allocation
 // targets. This includes dead nodes, but not decommissioning or decommissioned
 // nodes.

From ec55fa47f8ff6a1dfdfc2946730ae5fdcf77bed5 Mon Sep 17 00:00:00 2001
From: Rain <rain@oxide.computer>
Date: Mon, 6 Jul 2026 13:42:07 -0700
Subject: [PATCH 4/4] refactor policy floor into pure, unit-tested helpers
 (omicron#10658)

No behavior change. The floor arithmetic was duplicated inline in
computeAction and calcRangeCounter with two subtly different exclusion
predicates, and could only be tested through the full allocator and metrics
harnesses. Two copies of logic that must agree, drifting independently, is
the same failure mode this incident is about, so consolidate:

allocator_policyfloor.go now holds the whole concept:

- policyFloorNeededVoters: the floor arithmetic as a pure function of
  (needed, have, policyRemoved, configured).
- nodeLivenessStatusIsPolicyRemoved: the allocator-side predicate, an
  exhaustive switch over the status enum with a fail-safe default.
- policyRemovedVoterCount: the metric-side count for IsLiveMap consumers,
  with the deliberate allocator/gauge divergence for
  unavailable-but-not-yet-dead decommissioning nodes documented in one place.

computeAction, calcRangeCounter, and the StorePool helper reduce to calls
into these; the long inline rationale comments move to the helpers.

New unit tests, no cluster harness required:

- TestPolicyFloorNeededVoters: named scenarios (madrid, decommission,
  dead-node decommission, even state, RF lowered, over-RF).
- TestPolicyFloorNeededVotersInvariants: exhaustive enumeration over the
  input space (replica counts and RFs live in [0, 7], so full coverage is a
  few thousand cases and beats sampling): the floor never lowers the target,
  never exceeds max(needed, configured RF), never invents replicas, is
  monotone in policy removals, idempotent, and never sizes a healthy in-RF
  range below the replicas it already has (the madrid invariant).
- TestNodeLivenessStatusIsPolicyRemoved: exhaustive over the enum via its
  generated name map, so a newly added status fails the test and forces an
  explicit policy classification instead of silently inheriting the default.
- TestPolicyRemovedVoterCount: membership/absence cases for the gauge path.

Verified unchanged behavior: TestAllocator*, TestStorePool*, *Decommission*,
TestReplicateQueue*, *Metrics* all pass unmodified; go vet clean.
---
 pkg/kv/kvserver/allocator.go                  |  48 ++----
 pkg/kv/kvserver/allocator_policyfloor.go      | 111 +++++++++++++
 pkg/kv/kvserver/allocator_policyfloor_test.go | 147 ++++++++++++++++++
 pkg/kv/kvserver/replica_metrics.go            |  36 ++---
 pkg/kv/kvserver/store_pool.go                 |  24 +--
 5 files changed, 291 insertions(+), 75 deletions(-)
 create mode 100644 pkg/kv/kvserver/allocator_policyfloor.go

diff --git a/pkg/kv/kvserver/allocator.go b/pkg/kv/kvserver/allocator.go
index ebad54a1d8..dc2c5b76f0 100644
--- a/pkg/kv/kvserver/allocator.go
+++ b/pkg/kv/kvserver/allocator.go
@@ -690,39 +690,23 @@ func (a *Allocator) computeAction(
 	clusterNodes := a.storePool.ClusterNodeCount()
 	neededVoters := GetNeededVoters(conf.GetNumVoters(), clusterNodes)
 
-	// TODO-RAINCLAUDE: experiment 2 for omicron#10658 — policy-floor the effective
-	// RF. The effective replication factor must be driven exclusively by operator
-	// policy: the configured RF (conf.GetNumVoters()) and operator decommissioning.
-	// It must NOT be driven by transient cluster health. GetNeededVoters above
-	// derives neededVoters from clusterNodes, which is the leaseholder's cache-based
-	// node count and can read phantom-low when the liveness cache is cold (the
-	// omicron#10658 trigger). We therefore floor neededVoters at the number of this
-	// range's own voters that sit on nodes the operator has NOT directed out of the
-	// cluster, capped at the configured RF. The excluded set is keyed on liveness
-	// MEMBERSHIP (DECOMMISSIONING or DECOMMISSIONED), not on the decommissioningVoters
-	// status set above: a dead node under decommission reports DECOMMISSIONED, which
-	// classifies as storeStatusDead, and it must still lower the floor — otherwise
-	// its replica could only ever be replaced, never shed, and on a cluster with no
-	// spare node the decommission would stall in purgatory forever. A
-	// dead-but-not-decommissioned voter still counts toward the floor, so a cold
-	// cache can never size a healthy range below the replicas it already has and
-	// trim it; only lowering num_replicas or decommissioning (both operator policy)
-	// can reduce the effective RF. Safety of the subtraction: whenever it drops the
-	// floor below haveVoters, the excluded replicas are by construction either
-	// live-decommissioning (handled by the RemoveDecommissioningVoter branch below)
-	// or dead (handled by RemoveDeadVoter, whose removal candidates the replicate
-	// queue restricts to dead replicas), so the liveness-blind RemoveVoter branch
-	// still fires only when haveVoters exceeds the configured RF. The range
-	// descriptor is authoritative (Raft-replicated), so this needs no KV scan.
+	// TODO-RAINCLAUDE: experiment 2 for omicron#10658 — apply the policy floor
+	// to the allocator's target; see allocator_policyfloor.go for the rule and
+	// the helpers' invariants. clusterNodes above is the leaseholder's
+	// cache-based node count and can read phantom-low when the liveness cache
+	// is cold (the omicron#10658 trigger); the floor keeps a transient count
+	// from sizing a range below the voters it already holds on nodes the
+	// operator has not decommissioned, so the liveness-blind RemoveVoter branch
+	// below fires only when haveVoters exceeds the configured RF. The excluded
+	// set deliberately differs from decommissioningVoters above (status-keyed,
+	// live nodes only): a policy-removed voter is either live-decommissioning
+	// (handled by the RemoveDecommissioningVoter branch) or dead (handled by
+	// RemoveDeadVoter, whose removal candidates the replicate queue restricts
+	// to dead replicas). The range descriptor is authoritative
+	// (Raft-replicated), so this needs no KV scan.
 	policyRemovedVoters := a.storePool.decommissioningOrDecommissionedReplicas(voterReplicas)
-	if policyFloor := haveVoters - len(policyRemovedVoters); policyFloor > neededVoters {
-		if maxRF := int(conf.GetNumVoters()); policyFloor > maxRF {
-			policyFloor = maxRF
-		}
-		if policyFloor > neededVoters {
-			neededVoters = policyFloor
-		}
-	}
+	neededVoters = policyFloorNeededVoters(
+		neededVoters, haveVoters, len(policyRemovedVoters), int(conf.GetNumVoters()))
 
 	desiredQuorum := computeQuorum(neededVoters)
 	quorum := computeQuorum(haveVoters)
diff --git a/pkg/kv/kvserver/allocator_policyfloor.go b/pkg/kv/kvserver/allocator_policyfloor.go
new file mode 100644
index 0000000000..22d57fa1fe
--- /dev/null
+++ b/pkg/kv/kvserver/allocator_policyfloor.go
@@ -0,0 +1,111 @@
+// Copyright 2014 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package kvserver
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+)
+
+// TODO-RAINCLAUDE: omicron#10658 — the policy floor, in one place. The rule:
+// the effective replication factor is a function of operator policy alone
+// (the configured num_replicas and operator decommissioning), never of
+// transient cluster health (a dead node, a cold liveness cache, a phantom-low
+// node count). computeAction applies the floor to the allocator's target and
+// calcRangeCounter applies it to the under-/over-replication gauges, both
+// through the pure helpers below so the two sites cannot drift and the logic
+// is directly unit-testable (see the TestPolicyFloor* tests).
+
+// TODO-RAINCLAUDE: the floor arithmetic. Returns neededVoters floored at the
+// range's own voters on nodes the operator has not directed out of the
+// cluster (haveVoters - policyRemovedVoters), capped at the configured RF.
+// Pure; TestPolicyFloorNeededVotersInvariants proves the following properties
+// by exhaustive enumeration:
+//
+//   - The result never drops below neededVoters, so the downshift's
+//     small-cluster and decommission-completion behavior is preserved.
+//   - The result never exceeds max(neededVoters, configuredVoters), so
+//     lowering num_replicas still down-replicates.
+//   - The result never exceeds max(neededVoters, haveVoters -
+//     policyRemovedVoters): the floor cannot invent replicas.
+//   - With no policy-removed voters and haveVoters within the configured RF,
+//     the result is at least haveVoters — the madrid invariant: no phantom-low
+//     node count can size a healthy range below the replicas it already has,
+//     which is what authorized the omicron#10658 trim.
+func policyFloorNeededVoters(
+	neededVoters, haveVoters, policyRemovedVoters, configuredVoters int,
+) int {
+	policyFloor := haveVoters - policyRemovedVoters
+	if policyFloor > configuredVoters {
+		policyFloor = configuredVoters
+	}
+	if policyFloor > neededVoters {
+		return policyFloor
+	}
+	return neededVoters
+}
+
+// TODO-RAINCLAUDE: the allocator-side exclusion predicate: is this node's
+// liveness status the result of operator policy (decommissioning), such that
+// its replica should not count toward the policy floor? True for exactly
+// DECOMMISSIONING (live + non-active membership) and DECOMMISSIONED (dead +
+// non-active membership; this is what a dead node under decommission reports,
+// and it classifies as storeStatusDead — which is why the floor cannot key on
+// storeStatusDecommissioning). Everything else is false, and false is always
+// the fail-safe direction: an unexcluded replica keeps the floor high, which
+// can pause a decommission but can never remove a live replica. In
+// particular: UNKNOWN (no liveness record — the cold-cache state that
+// triggered omicron#10658) and UNAVAILABLE (expired but not yet past the dead
+// threshold; LivenessStatus reports this even for a decommissioning node, and
+// excluding it here would open a liveness-blind RemoveVoter window in
+// computeAction while the node sits in neither the dead nor the
+// decommissioning status set). The default arm makes any future status
+// fail-safe too; TestNodeLivenessStatusIsPolicyRemoved forces an explicit
+// classification when a status is added.
+func nodeLivenessStatusIsPolicyRemoved(status livenesspb.NodeLivenessStatus) bool {
+	switch status {
+	case livenesspb.NodeLivenessStatus_DECOMMISSIONING,
+		livenesspb.NodeLivenessStatus_DECOMMISSIONED:
+		return true
+	case livenesspb.NodeLivenessStatus_UNKNOWN,
+		livenesspb.NodeLivenessStatus_DEAD,
+		livenesspb.NodeLivenessStatus_UNAVAILABLE,
+		livenesspb.NodeLivenessStatus_LIVE,
+		livenesspb.NodeLivenessStatus_DRAINING:
+		return false
+	default:
+		return false
+	}
+}
+
+// TODO-RAINCLAUDE: the metric-side exclusion count for calcRangeCounter,
+// which has an IsLiveMap rather than a NodeLivenessFunc. Keyed on raw
+// liveness membership because the map carries no dead-threshold
+// classification; this diverges from nodeLivenessStatusIsPolicyRemoved only
+// while a decommissioning node is unavailable-but-not-yet-dead (the allocator
+// pauses; the gauge already treats the node as policy-removed), which is
+// acceptable for an estimated gauge. A voter absent from the map does not
+// count as removed — an unverifiable replica must not lower the target — so
+// during a cold-cache window a healthy range reads as under-replicated (the
+// honest signal that the leaseholder cannot account for its replicas) rather
+// than over-replicated.
+func policyRemovedVoterCount(
+	voters []roachpb.ReplicaDescriptor, livenessMap liveness.IsLiveMap,
+) int {
+	removed := 0
+	for _, rd := range voters {
+		if entry, ok := livenessMap[rd.NodeID]; ok && !entry.Membership.Active() {
+			removed++
+		}
+	}
+	return removed
+}
diff --git a/pkg/kv/kvserver/allocator_policyfloor_test.go b/pkg/kv/kvserver/allocator_policyfloor_test.go
index 0037f356fb..8bfcc755f5 100644
--- a/pkg/kv/kvserver/allocator_policyfloor_test.go
+++ b/pkg/kv/kvserver/allocator_policyfloor_test.go
@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -318,3 +319,149 @@ func TestAllocatorPolicyFloorDeadNodeDecommission(t *testing.T) {
 		})
 	}
 }
+
+// TODO-RAINCLAUDE: unit tests for the pure helpers in
+// allocator_policyfloor.go. Named scenarios below; the invariants are proved
+// separately by exhaustive enumeration in
+// TestPolicyFloorNeededVotersInvariants.
+func TestPolicyFloorNeededVoters(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	cases := []struct {
+		name                                          string
+		needed, have, policyRemoved, configured, want int
+	}{
+		{"madrid: healthy RF-5 range, phantom count downshifted needed to 3", 3, 5, 0, 5, 5},
+		{"small cluster: RF-5 range with 3 voters on 3 nodes", 3, 3, 0, 5, 3},
+		{"under-replicated: the floor never lowers the target", 5, 3, 0, 5, 5},
+		{"live decommission 2 of 5, phantom count", 3, 5, 2, 5, 3},
+		{"dead-node decommission 1 of 5", 3, 5, 1, 5, 4},
+		{"even steady state: 4 healthy voters on 4 nodes", 3, 4, 0, 5, 4},
+		{"over-replicated beyond the configured RF", 5, 6, 0, 5, 5},
+		{"over-replicated beyond the configured RF, phantom count", 3, 6, 0, 5, 5},
+		{"num_replicas lowered 5 -> 3: the cap lets the trim proceed", 3, 5, 0, 3, 3},
+		{"all voters policy-removed", 3, 5, 5, 5, 3},
+		{"no voters", 3, 0, 0, 5, 3},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			require.Equal(t, c.want,
+				policyFloorNeededVoters(c.needed, c.have, c.policyRemoved, c.configured))
+		})
+	}
+}
+
+// TODO-RAINCLAUDE: the floor's invariants, proved by exhaustive enumeration.
+// Real replica counts and RFs live in [0, 7], so full coverage of the
+// meaningful input space is a few thousand cases — strictly stronger than
+// sampling and dependency-free.
+func TestPolicyFloorNeededVotersInvariants(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	const max = 7
+	for needed := 0; needed <= max; needed++ {
+		for have := 0; have <= max; have++ {
+			for removed := 0; removed <= have; removed++ {
+				for configured := 0; configured <= max; configured++ {
+					got := policyFloorNeededVoters(needed, have, removed, configured)
+					args := []interface{}{needed, have, removed, configured}
+
+					// The floor only ever raises the target: the downshift's
+					// small-cluster and decommission behavior is preserved.
+					require.GreaterOrEqualf(t, got, needed,
+						"floor lowered the target (needed=%d have=%d removed=%d configured=%d)", args...)
+					if got > needed {
+						// When it raises, it raises to no more than the configured RF
+						// (lowering num_replicas still down-replicates)...
+						require.LessOrEqualf(t, got, configured,
+							"floor exceeded the configured RF (needed=%d have=%d removed=%d configured=%d)", args...)
+						// ...and to no more than the voters actually held on
+						// policy-active nodes (the floor cannot invent replicas).
+						require.LessOrEqualf(t, got, have-removed,
+							"floor invented replicas (needed=%d have=%d removed=%d configured=%d)", args...)
+					}
+					// The madrid invariant: with no policy-removed voters and
+					// haveVoters within the configured RF, a range is never sized
+					// below the replicas it already has — no phantom-low node count
+					// can authorize a trim.
+					if removed == 0 && have <= configured {
+						require.GreaterOrEqualf(t, got, have,
+							"healthy in-RF range sized below itself (needed=%d have=%d removed=%d configured=%d)", args...)
+					}
+					// Monotone: one more policy-removed voter never raises the target.
+					if removed < have {
+						require.GreaterOrEqualf(t, got,
+							policyFloorNeededVoters(needed, have, removed+1, configured),
+							"extra policy removal raised the target (needed=%d have=%d removed=%d configured=%d)", args...)
+					}
+					// Idempotent: re-applying the floor changes nothing.
+					require.Equalf(t, got, policyFloorNeededVoters(got, have, removed, configured),
+						"floor is not idempotent (needed=%d have=%d removed=%d configured=%d)", args...)
+				}
+			}
+		}
+	}
+}
+
+// TODO-RAINCLAUDE: exhaustive over the NodeLivenessStatus enum via its
+// generated name map, so adding a status to the proto forces an explicit
+// policy-floor classification here instead of silently inheriting the
+// fail-safe default.
+func TestNodeLivenessStatusIsPolicyRemoved(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	want := map[livenesspb.NodeLivenessStatus]bool{
+		livenesspb.NodeLivenessStatus_UNKNOWN:         false, // no liveness record (cold cache): fail-safe
+		livenesspb.NodeLivenessStatus_DEAD:            false, // deadness is health, not policy
+		livenesspb.NodeLivenessStatus_UNAVAILABLE:     false, // excluding it would open a blind RemoveVoter window
+		livenesspb.NodeLivenessStatus_LIVE:            false,
+		livenesspb.NodeLivenessStatus_DECOMMISSIONING: true,  // operator policy, node live
+		livenesspb.NodeLivenessStatus_DECOMMISSIONED:  true,  // operator policy, node dead
+		livenesspb.NodeLivenessStatus_DRAINING:        false, // restart in progress, not removal
+	}
+
+	for value, name := range livenesspb.NodeLivenessStatus_name {
+		status := livenesspb.NodeLivenessStatus(value)
+		expected, ok := want[status]
+		require.Truef(t, ok,
+			"NodeLivenessStatus %s has no policy-floor classification; decide whether it represents operator removal",
+			name)
+		require.Equalf(t, expected, nodeLivenessStatusIsPolicyRemoved(status), "status %s", name)
+	}
+	require.Len(t, want, len(livenesspb.NodeLivenessStatus_name))
+}
+
+func TestPolicyRemovedVoterCount(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	voters := []roachpb.ReplicaDescriptor{
+		{NodeID: 1, StoreID: 10, ReplicaID: 1},
+		{NodeID: 2, StoreID: 20, ReplicaID: 2},
+		{NodeID: 3, StoreID: 30, ReplicaID: 3},
+		{NodeID: 4, StoreID: 40, ReplicaID: 4},
+		{NodeID: 5, StoreID: 50, ReplicaID: 5},
+	}
+	livenessMap := liveness.IsLiveMap{
+		// Node 1 is deliberately absent from the map: an unverifiable replica
+		// counts toward the floor, not as removed.
+		2: {IsLive: true},  // membership active, live
+		3: {IsLive: false}, // membership active, dead: health, not policy
+		4: { // live decommissioning
+			Liveness: livenesspb.Liveness{Membership: livenesspb.MembershipStatus_DECOMMISSIONING},
+			IsLive:   true,
+		},
+		5: { // dead and fully decommissioned
+			Liveness: livenesspb.Liveness{Membership: livenesspb.MembershipStatus_DECOMMISSIONED},
+			IsLive:   false,
+		},
+	}
+
+	require.Equal(t, 2, policyRemovedVoterCount(voters, livenessMap))
+	require.Equal(t, 0, policyRemovedVoterCount(nil, livenessMap))
+	require.Equal(t, 0, policyRemovedVoterCount(voters, liveness.IsLiveMap{}))
+}
diff --git a/pkg/kv/kvserver/replica_metrics.go b/pkg/kv/kvserver/replica_metrics.go
index 1213abd76a..07be581a29 100644
--- a/pkg/kv/kvserver/replica_metrics.go
+++ b/pkg/kv/kvserver/replica_metrics.go
@@ -181,33 +181,15 @@ func calcRangeCounter(
 	if rangeCounter {
 		neededVoters := GetNeededVoters(numVoters, clusterNodes)
 		// TODO-RAINCLAUDE: omicron#10658 — the same policy floor computeAction
-		// applies, so the gauges agree with the allocator. Without it, a phantom-low
-		// clusterNodes makes ranges_underreplicated go blind exactly when it matters
-		// and false-alarms ranges_overreplicated on every healthy RF-5 range; a
-		// cluster legitimately settled above the downshifted target (e.g. 4 voters
-		// on 4 nodes after a decommission) reads over-replicated forever. The floor
-		// counts this range's voters on membership-active nodes, capped at the
-		// configured RF. A voter absent from the liveness map counts toward the
-		// floor (fail-safe: an unverifiable replica must not lower the target), so
-		// during a cold-cache window a healthy range reads as under-replicated —
-		// the honest signal that the leaseholder cannot account for its replicas —
-		// rather than over-replicated. Metric-only: keyed on raw membership, which
-		// may diverge from the allocator's floor for the brief window where a
-		// decommissioning node is unavailable but not yet dead.
-		voterFloor := 0
-		for _, rd := range desc.Replicas().VoterDescriptors() {
-			if entry, ok := livenessMap[rd.NodeID]; !ok || entry.Membership.Active() {
-				voterFloor++
-			}
-		}
-		if voterFloor > neededVoters {
-			if maxRF := int(numVoters); voterFloor > maxRF {
-				voterFloor = maxRF
-			}
-			if voterFloor > neededVoters {
-				neededVoters = voterFloor
-			}
-		}
+		// applies, so the gauges agree with the allocator; see
+		// allocator_policyfloor.go. Without it, a phantom-low clusterNodes makes
+		// ranges_underreplicated go blind exactly when it matters and false-alarms
+		// ranges_overreplicated on every healthy RF-5 range, and states the
+		// allocator deliberately preserves (e.g. 4 voters on 4 nodes after a
+		// decommission) read over-replicated forever.
+		voterDescs := desc.Replicas().VoterDescriptors()
+		neededVoters = policyFloorNeededVoters(neededVoters, len(voterDescs),
+			policyRemovedVoterCount(voterDescs, livenessMap), int(numVoters))
 		neededNonVoters := GetNeededNonVoters(int(numVoters), int(numReplicas-numVoters), clusterNodes)
 		status := desc.Replicas().ReplicationStatus(func(rDesc roachpb.ReplicaDescriptor) bool {
 			return livenessMap[rDesc.NodeID].IsLive
diff --git a/pkg/kv/kvserver/store_pool.go b/pkg/kv/kvserver/store_pool.go
index 1e5d6656b7..a95da32210 100644
--- a/pkg/kv/kvserver/store_pool.go
+++ b/pkg/kv/kvserver/store_pool.go
@@ -598,19 +598,13 @@ func (sp *StorePool) decommissioningReplicas(
 	return
 }
 
-// TODO-RAINCLAUDE: fix for the omicron#10658 policy floor — the set of replicas
-// the operator has directed out of the cluster, keyed on liveness MEMBERSHIP
-// rather than store-pool status. decommissioningReplicas above only matches
-// storeStatusDecommissioning, which requires a LIVE decommissioning node: a
-// dead node under decommission reports NodeLivenessStatus_DECOMMISSIONED
-// (dead + non-active membership) and lands in storeStatusDead instead. The
-// policy floor must exclude both, or a dead node's replicas can never be shed.
-// An UNAVAILABLE node (expired but not yet past the dead threshold) is
-// deliberately NOT matched even if its membership is non-active: excluding it
-// from the floor while it sits in neither the dead nor the decommissioning
-// status set would open a liveness-blind RemoveVoter window in computeAction.
-// A node absent from the liveness cache reads as UNKNOWN and is likewise not
-// matched, which keeps the floor high — the fail-safe direction.
+// TODO-RAINCLAUDE: omicron#10658 policy floor — the replicas the operator has
+// directed out of the cluster, per nodeLivenessStatusIsPolicyRemoved (see
+// allocator_policyfloor.go for the predicate's rationale and fail-safe
+// properties). Distinct from decommissioningReplicas above, which keys on
+// storeStatusDecommissioning and therefore only matches LIVE decommissioning
+// nodes; a dead node under decommission classifies as storeStatusDead but is
+// still policy-removed.
 func (sp *StorePool) decommissioningOrDecommissionedReplicas(
 	repls []roachpb.ReplicaDescriptor,
 ) (inactive []roachpb.ReplicaDescriptor) {
@@ -618,9 +612,7 @@ func (sp *StorePool) decommissioningOrDecommissionedReplicas(
 	timeUntilStoreDead := TimeUntilStoreDead.Get(&sp.st.SV)
 
 	for _, repl := range repls {
-		switch sp.nodeLivenessFn(repl.NodeID, now, timeUntilStoreDead) {
-		case livenesspb.NodeLivenessStatus_DECOMMISSIONING,
-			livenesspb.NodeLivenessStatus_DECOMMISSIONED:
+		if nodeLivenessStatusIsPolicyRemoved(sp.nodeLivenessFn(repl.NodeID, now, timeUntilStoreDead)) {
 			inactive = append(inactive, repl)
 		}
 	}