asim: add ChangeReplicasOp as an operation

This commit adds ChangeReplicasOp, enabling the simulator’s store rebalancer to
enqueue replica change operations. It is currently unused; future commits will
integrate it with mma to enqueue replica changes.

Epic: none
Release note: none

Author: wenyihu6

pkg/kv/kvserver/asim/op/BUILD.bazel

@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "op",
     srcs = [
+        "change_replicas.go",
         "controller.go",
         "operation.go",
         "pq.go",
@@ -12,6 +13,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/op",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/allocator",
         "//pkg/kv/kvserver/allocator/allocatorimpl",

pkg/kv/kvserver/asim/op/change_replicas.go

@@ -0,0 +1,40 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package op
+
+import (
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/asim/state"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/errors"
+)
+
+// ChangeReplicasOp contains the information for a change replicas operation.
+type ChangeReplicasOp struct {
+	baseOp
+	rangeID state.RangeID
+	changes kvpb.ReplicationChanges
+}
+
+// NewChangeReplicasOp returns a new ChangeReplicasOp.
+// TODO(wenyihu6): unused for now - will be integrated with mma simulation
+func NewChangeReplicasOp(
+	tick time.Time, rangeID roachpb.RangeID, changes kvpb.ReplicationChanges,
+) *ChangeReplicasOp {
+	return &ChangeReplicasOp{
+		baseOp:  newBaseOp(tick),
+		rangeID: state.RangeID(rangeID),
+		changes: changes,
+	}
+}
+
+func (cro *ChangeReplicasOp) error(err error) {
+	augmentedErr := errors.Wrapf(err, "Unable to change replicas for range_id=%d, changes=%v",
+		cro.rangeID, cro.changes)
+	cro.errs = append(cro.errs, augmentedErr)
+}

pkg/kv/kvserver/asim/op/controller.go

@@ -134,6 +134,12 @@ func (c *controller) process(
 			op.done = true
 			op.complete = tick
 		}
+	case *ChangeReplicasOp:
+		if err := c.processChangeReplicas(tick, state, op); err != nil {
+			op.error(err)
+			op.done = true
+			op.complete = tick
+		}
 	default:
 		return
 	}
@@ -228,3 +234,46 @@ func (c *controller) processTransferLease(
 	ro.next = tick.Add(delay)
 	return nil
 }
+
+func (c *controller) processChangeReplicas(
+	tick time.Time, s state.State, cro *ChangeReplicasOp,
+) error {
+	rng, ok := s.Range(cro.rangeID)
+	if !ok {
+		panic(errors.Newf("programming error: range %d not found", cro.rangeID))
+	}
+	// We need to check if the change is already complete. If it is, we can
+	// skip the operation. This is the case where the change didn't apply
+	// instantly and processChangeReplicas is called over multiple ticks.
+	if (cro.complete != time.Time{}) && !tick.Before(cro.complete) {
+		cro.done = true
+		return nil
+	}
+
+	change := state.ReplicaChange{
+		RangeID: cro.rangeID,
+		Author:  c.storeID,
+		Changes: cro.changes,
+	}
+
+	if len(cro.changes) == 0 {
+		panic("unexpected empty changes in ChangeReplicasOp")
+	}
+
+	targets := kvserver.SynthesizeTargetsByChangeType(cro.changes)
+	change.Wait = c.settings.ReplicaChangeDelayFn()(rng.Size(),
+		len(targets.VoterAdditions) > 0 || len(targets.NonVoterAdditions) > 0)
+	completeAt, ok := c.changer.Push(tick, &change)
+	if !ok {
+		return errors.Newf("tick %d: Changer did not accept change for range %d", tick, cro.rangeID)
+	}
+	cro.complete = completeAt
+
+	if !tick.Before(completeAt) {
+		// If the change was applied instantly (only promotion/demotion).
+		cro.done = true
+	} else {
+		cro.next = completeAt
+	}
+	return nil
+}