cockroachdb
diff --git a/‎pkg/cli/declarative_print_rules.go‎
Lines changed: 1 addition & 1 deletion b/‎pkg/cli/declarative_print_rules.go‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎pkg/cli/testdata/declarative-rules/deprules‎
Lines changed: 266 additions & 250 deletions b/‎pkg/cli/testdata/declarative-rules/deprules‎
Lines changed: 266 additions & 250 deletions
diff --git a/‎pkg/cli/testdata/declarative-rules/invalid_version‎
Lines changed: 2 additions & 2 deletions b/‎pkg/cli/testdata/declarative-rules/invalid_version‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎pkg/kv/kvserver/batcheval/cmd_end_transaction.go‎
Lines changed: 3 additions & 0 deletions b/‎pkg/kv/kvserver/batcheval/cmd_end_transaction.go‎
Lines changed: 3 additions & 0 deletions
diff --git a/‎pkg/kv/kvserver/kvstorage/BUILD.bazel‎
Lines changed: 2 additions & 0 deletions b/‎pkg/kv/kvserver/kvstorage/BUILD.bazel‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎pkg/kv/kvserver/kvstorage/datadriven_test.go‎
Lines changed: 15 additions & 0 deletions b/‎pkg/kv/kvserver/kvstorage/datadriven_test.go‎
Lines changed: 15 additions & 0 deletions
diff --git a/‎pkg/kv/kvserver/kvstorage/init.go‎
Lines changed: 115 additions & 91 deletions b/‎pkg/kv/kvserver/kvstorage/init.go‎
Lines changed: 115 additions & 91 deletions
@@ -37,7 +37,7 @@ a given corpus file.
 					Version: version,
 				})
 			if rules == nil {
-				fmt.Printf("unsupported version number, the supported versions are: \n")
+				fmt.Printf("unsupported version number, the supported versions are:\n")
 				for _, v := range scplan.GetReleasesForRulesRegistries() {
 					fmt.Printf(" %s\n", v)
 				}
 
@@ -2,7 +2,7 @@ invalid_version
 -----
 ----
 debug declarative-print-rules 1.1 op
-unsupported version number, the supported versions are: 
+unsupported version number, the supported versions are:
  latest
- 1000025.2
  1000025.3
+ 1000025.2
@@ -1407,6 +1407,9 @@ func splitTriggerHelper(
 	// This avoids running the consistency checker on the RHS immediately after
 	// the split.
 	lastTS := hlc.Timestamp{}
+	// TODO(arul): instead of fetching the consistency checker timestamp here
+	// like this, we should instead pass it using the SplitTriggerHelperInput to
+	// make it easier to test.
 	if _, err := storage.MVCCGetProto(ctx, batch,
 		keys.QueueLastProcessedKey(split.LeftDesc.StartKey, "consistencyChecker"),
 		hlc.Timestamp{}, &lastTS, storage.MVCCGetOptions{}); err != nil {
 
@@ -42,6 +42,7 @@ go_test(
         "cluster_version_test.go",
         "datadriven_test.go",
         "destroy_test.go",
+        "init_test.go",
         "initial_test.go",
         "stateloader_test.go",
     ],
@@ -66,6 +67,7 @@ go_test(
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
+        "//pkg/util/randutil",
         "//pkg/util/stop",
         "//pkg/util/tracing",
         "//pkg/util/uuid",
 
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverpb"
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
@@ -94,6 +95,14 @@ func (e *env) handleNewReplica(
 	return desc
 }
 
+func (e *env) handleRangeTombstone(
+	t *testing.T, ctx context.Context, rangeID roachpb.RangeID, next roachpb.ReplicaID,
+) {
+	require.NoError(t, MakeStateLoader(rangeID).SetRangeTombstone(
+		ctx, e.eng, kvserverpb.RangeTombstone{NextReplicaID: next},
+	))
+}
+
 func TestDataDriven(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -146,6 +155,12 @@ func TestDataDriven(t *testing.T) {
 				); desc != nil {
 					fmt.Fprintln(&buf, desc)
 				}
+
+			case "range-tombstone":
+				rangeID := dd.ScanArg[roachpb.RangeID](t, d, "range-id")
+				nextID := dd.ScanArg[roachpb.ReplicaID](t, d, "next-replica-id")
+				e.handleRangeTombstone(t, ctx, rangeID, nextID)
+
 			case "load-and-reconcile":
 				replicas, err := LoadAndReconcileReplicas(ctx, e.eng)
 				if err != nil {
 
@@ -9,6 +9,7 @@ import (
 	"bytes"
 	"cmp"
 	"context"
+	"maps"
 	"slices"
 	"time"
 
@@ -144,21 +145,27 @@ func checkCanInitializeEngine(ctx context.Context, eng storage.Engine) error {
 	return err
 }
 
-// IterateIDPrefixKeys helps visit system keys that use RangeID prefixing (such
+// readKeyFn reads the given key, and unmarshals the value into the given proto.
+// Returns false if the key does not exist, or is requested out of order.
+type readKeyFn func(roachpb.Key, protoutil.Message) (bool, error)
+
+// scanRangeIDFn reports the existence of a RangeID, and allows reading
+// RangeID-local keys via the readKeyFn callback.
+type scanRangeIDFn func(roachpb.RangeID, readKeyFn) error
+
+// iterateRangeIDKeys helps visit storage keys that use RangeID prefixing (such
 // as RaftHardStateKey, RangeTombstoneKey, and many others). Such keys could in
-// principle exist at any RangeID, and this helper efficiently discovers all the
-// keys of the desired type (as specified by the supplied `keyFn`) and, for each
-// key-value pair discovered, unmarshals it into `msg` and then invokes `f`.
+// principle exist for any RangeID.
+//
+// The helper visits all RangeIDs that have any keys, and for each range calls
+// the scanRangeID function. The implementation of this function can request any
+// subset of RangeID-local keys via the readKeyFn callback. All keys must be
+// requested in sorted order.
 //
 // Iteration stops on the first error (and will pass through that error).
-func IterateIDPrefixKeys(
-	ctx context.Context,
-	reader storage.Reader,
-	keyFn func(roachpb.RangeID) roachpb.Key,
-	msg protoutil.Message,
-	f func(_ roachpb.RangeID) error,
+func iterateRangeIDKeys(
+	ctx context.Context, reader storage.Reader, scanRangeID scanRangeIDFn,
 ) error {
-	rangeID := roachpb.RangeID(1)
 	// NB: Range-ID local keys have no versions and no intents.
 	iter, err := reader.NewMVCCIterator(ctx, storage.MVCCKeyIterKind, storage.IterOptions{
 		UpperBound: keys.LocalRangeIDPrefix.PrefixEnd().AsRawKey(),
@@ -168,60 +175,62 @@ func IterateIDPrefixKeys(
 	}
 	defer iter.Close()
 
-	for {
-		bumped := false
-		mvccKey := storage.MakeMVCCMetadataKey(keyFn(rangeID))
-		iter.SeekGE(mvccKey)
+	iter.SeekGE(storage.MakeMVCCMetadataKey(keys.LocalRangeIDPrefix.AsRawKey()))
+	iterOK, iterErr := iter.Valid()
+	if !iterOK || iterErr != nil {
+		return iterErr
+	}
 
-		if ok, err := iter.Valid(); !ok {
-			return err
+	getKeyFn := func(key roachpb.Key, msg protoutil.Message) (bool, error) {
+		if !iterOK || iterErr != nil {
+			return iterOK, iterErr
 		}
-
-		unsafeKey := iter.UnsafeKey()
-
-		if !bytes.HasPrefix(unsafeKey.Key, keys.LocalRangeIDPrefix) {
-			// Left the local keyspace, so we're done.
-			return nil
+		unsafeKey := iter.UnsafeKey().Key
+		comp := unsafeKey.Compare(key)
+		if comp < 0 {
+			iter.SeekGE(storage.MakeMVCCMetadataKey(key))
+			if iterOK, iterErr = iter.Valid(); !iterOK || iterErr != nil {
+				return iterOK, iterErr
+			}
+			unsafeKey = iter.UnsafeKey().Key
+			comp = unsafeKey.Compare(key)
+			if comp < 0 {
+				return false, errors.AssertionFailedf("SeekGE undershot key %s", key)
+			}
 		}
-
-		curRangeID, _, _, _, err := keys.DecodeRangeIDKey(unsafeKey.Key)
-		if err != nil {
-			return err
+		if comp > 0 {
+			return false, nil
 		}
-
-		if curRangeID > rangeID {
-			// `bumped` is always `false` here, but let's be explicit.
-			if !bumped {
-				rangeID = curRangeID
-				bumped = true
-			}
-			mvccKey = storage.MakeMVCCMetadataKey(keyFn(rangeID))
+		// Found the key (comp == 0). Parse and report the value.
+		var meta enginepb.MVCCMetadata
+		if err := iter.ValueProto(&meta); err != nil {
+			return false, errors.Errorf("unable to unmarshal %s into MVCCMetadata", unsafeKey)
 		}
-
-		if !unsafeKey.Key.Equal(mvccKey.Key) {
-			if !bumped {
-				// Don't increment the rangeID if it has already been incremented
-				// above, or we could skip past a value we ought to see.
-				rangeID++
-				bumped = true // for completeness' sake; continuing below anyway
-			}
-			continue
+		val := roachpb.Value{RawBytes: meta.RawBytes}
+		if err := val.GetProto(msg); err != nil {
+			return false, errors.Errorf("unable to unmarshal %s into %T", unsafeKey, msg)
 		}
+		return true, nil
+	}
 
-		ok, err := storage.MVCCGetProto(
-			ctx, reader, unsafeKey.Key, hlc.Timestamp{}, msg, storage.MVCCGetOptions{})
+	for iterOK && iterErr == nil {
+		rangeID, _, _, _, err := keys.DecodeRangeIDKey(iter.UnsafeKey().Key)
 		if err != nil {
 			return err
-		}
-		if !ok {
-			return errors.Errorf("unable to unmarshal %s into %T", unsafeKey.Key, msg)
-		}
-
-		if err := f(rangeID); err != nil {
+		} else if err := scanRangeID(rangeID, getKeyFn); err != nil {
 			return iterutil.Map(err)
+		} else if !iterOK || iterErr != nil {
+			return iterErr
+		}
+		newRangeID, _, _, _, err := keys.DecodeRangeIDKey(iter.UnsafeKey().Key)
+		if err != nil {
+			return err
+		} else if newRangeID <= rangeID {
+			iter.SeekGE(storage.MakeMVCCMetadataKey(keys.MakeRangeIDPrefix(rangeID + 1)))
+			iterOK, iterErr = iter.Valid()
 		}
-		rangeID++
 	}
+	return iterErr
 }
 
 // ReadStoreIdent reads the StoreIdent from the store.
@@ -408,6 +417,7 @@ type Replica struct {
 	ReplicaID roachpb.ReplicaID
 	Desc      *roachpb.RangeDescriptor // nil for uninitialized Replica
 
+	tombstone kvserverpb.RangeTombstone
 	hardState raftpb.HardState // internal to kvstorage, see migration in LoadAndReconcileReplicas
 }
 
@@ -454,9 +464,12 @@ func (m replicaMap) getOrMake(rangeID roachpb.RangeID) Replica {
 	return ent
 }
 
-func (m replicaMap) setReplicaID(rangeID roachpb.RangeID, replicaID roachpb.ReplicaID) {
+func (m replicaMap) setReplicaIDAndTombstone(
+	rangeID roachpb.RangeID, replicaID roachpb.ReplicaID, ts kvserverpb.RangeTombstone,
+) {
 	ent := m.getOrMake(rangeID)
 	ent.ReplicaID = replicaID
+	ent.tombstone = ts
 	m[rangeID] = ent
 }
 
@@ -499,52 +512,57 @@ func loadReplicas(ctx context.Context, eng storage.Engine) ([]Replica, error) {
 		}
 	}
 
-	// Load replicas from disk based on their RaftReplicaID and HardState.
+	// Scan all RangeIDs present on the store, and load ReplicaID and HardState of
+	// those that correspond to an existing replica (uninitialized or initialized).
+	//
+	// INVARIANT: a RangeID with no replica can only have a RangeTombstone key.
+	// INVARIANT: all replicas have a persisted ReplicaID.
+	// INVARIANT: ReplicaID >= RangeTombstone.NextReplicaID.
 	//
-	// INVARIANT: all replicas have a persisted full ReplicaID (i.e. a "ReplicaID from disk").
+	// NB: RangeIDs that only have a RangeTombstone are effectively skipped here
+	// as uninteresting. If a non-RangeTombstone key is found for a RangeID, there
+	// must be a replica of this range, so we load it and check invariants later.
 	//
 	// TODO(tbg): tighten up the case where we see a RaftReplicaID but no HardState.
 	// This leads to the general desire to validate the internal consistency of the
 	// entire raft state (i.e. HardState, TruncatedState, Log).
-	{
-		logEvery := log.Every(10 * time.Second)
-		var i int
-		var msg kvserverpb.RaftReplicaID
-		if err := IterateIDPrefixKeys(ctx, eng, func(rangeID roachpb.RangeID) roachpb.Key {
-			return keys.RaftReplicaIDKey(rangeID)
-		}, &msg, func(rangeID roachpb.RangeID) error {
-			if logEvery.ShouldLog() && i > 0 { // only log if slow
-				log.KvExec.Infof(ctx, "loaded replica ID for %d/%d replicas", i, len(s))
-			}
-			i++
-			s.setReplicaID(rangeID, msg.ReplicaID)
-			return nil
-		}); err != nil {
-			return nil, err
+	logEvery := log.Every(10 * time.Second)
+	var i int
+	if err := iterateRangeIDKeys(ctx, eng, func(id roachpb.RangeID, get readKeyFn) error {
+		if logEvery.ShouldLog() && i > 0 { // only log if slow
+			log.KvExec.Infof(ctx, "loaded state for %d/%d replicas", i, len(s))
 		}
-		log.KvExec.Infof(ctx, "loaded replica ID for %d/%d replicas", len(s), len(s))
-
-		logEvery = log.Every(10 * time.Second)
-		i = 0
+		i++
+		// NB: the keys must be requested in sorted order here.
+		buf := keys.MakeRangeIDPrefixBuf(id)
+		var ts kvserverpb.RangeTombstone
+		if ok, err := get(buf.RangeTombstoneKey(), &ts); err != nil {
+			return err
+		} else if !ok {
+			ts = kvserverpb.RangeTombstone{} // just in case it was mutated
+		}
+		// NB: the keys must be requested in sorted order here.
 		var hs raftpb.HardState
-		if err := IterateIDPrefixKeys(ctx, eng, func(rangeID roachpb.RangeID) roachpb.Key {
-			return keys.RaftHardStateKey(rangeID)
-		}, &hs, func(rangeID roachpb.RangeID) error {
-			if logEvery.ShouldLog() && i > 0 { // only log if slow
-				log.KvExec.Infof(ctx, "loaded Raft state for %d/%d replicas", i, len(s))
-			}
-			i++
-			s.setHardState(rangeID, hs)
-			return nil
-		}); err != nil {
-			return nil, err
+		if ok, err := get(buf.RaftHardStateKey(), &hs); err != nil {
+			return err
+		} else if ok {
+			s.setHardState(id, hs)
 		}
-		log.KvExec.Infof(ctx, "loaded Raft state for %d/%d replicas", len(s), len(s))
-	}
-	sl := make([]Replica, 0, len(s))
-	for _, repl := range s {
-		sl = append(sl, repl)
+		// NB: the keys must be requested in sorted order here.
+		var rID kvserverpb.RaftReplicaID
+		if ok, err := get(buf.RaftReplicaIDKey(), &rID); err != nil {
+			return err
+		} else if ok {
+			// NB: the tombstone can be empty.
+			s.setReplicaIDAndTombstone(id, rID.ReplicaID, ts)
+		}
+		return nil
+	}); err != nil {
+		return nil, err
 	}
+	log.KvExec.Infof(ctx, "loaded state for %d/%d replicas", len(s), len(s))
+
+	sl := slices.AppendSeq(make([]Replica, 0, len(s)), maps.Values(s))
 	slices.SortFunc(sl, func(a, b Replica) int {
 		return cmp.Compare(a.RangeID, b.RangeID)
 	})
@@ -583,6 +601,12 @@ func LoadAndReconcileReplicas(ctx context.Context, eng storage.Engine) ([]Replic
 		if repl.ReplicaID == 0 {
 			return nil, errors.AssertionFailedf("no RaftReplicaID for %s", repl.Desc)
 		}
+		// INVARIANT: ReplicaID >= RangeTombstone.NextReplicaID.
+		if repl.ReplicaID < repl.tombstone.NextReplicaID {
+			return nil, errors.AssertionFailedf(
+				"r%d: RaftReplicaID %d survived RangeTombstone %+v",
+				repl.RangeID, repl.ReplicaID, repl.tombstone)
+		}
 
 		if repl.Desc != nil {
 			// INVARIANT: a Replica's RangeDescriptor always contains the local Store,
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ a given corpus file.`
`37`	`37`	`Version: version,`
`38`	`38`	`})`
`39`	`39`	`if rules == nil {`
`40`		`- fmt.Printf("unsupported version number, the supported versions are: \n")`
	`40`	`+ fmt.Printf("unsupported version number, the supported versions are:\n")`
`41`	`41`	`for _, v := range scplan.GetReleasesForRulesRegistries() {`
`42`	`42`	`fmt.Printf(" %s\n", v)`
`43`	`43`	`}`