kvserver: use FullReplicaID in creation stack

Epic: none
Release note: none

Author: pav-kv

pkg/kv/kvserver/kvstorage/replica_state.go

@@ -71,6 +71,10 @@ func LoadReplicaState(
 	return ls, nil
 }
 
+func (r LoadedReplicaState) FullReplicaID() storage.FullReplicaID {
+	return storage.FullReplicaID{RangeID: r.ReplState.Desc.RangeID, ReplicaID: r.ReplicaID}
+}
+
 // check makes sure that the replica invariants hold for the loaded state.
 func (r LoadedReplicaState) check(storeID roachpb.StoreID) error {
 	desc := r.ReplState.Desc
@@ -116,20 +120,16 @@ const CreateUninitReplicaTODO = 0
 // Returns kvpb.RaftGroupDeletedError if this replica can not be created
 // because it has been deleted.
 func CreateUninitializedReplica(
-	ctx context.Context,
-	eng storage.Engine,
-	storeID roachpb.StoreID,
-	rangeID roachpb.RangeID,
-	replicaID roachpb.ReplicaID,
+	ctx context.Context, eng storage.Engine, storeID roachpb.StoreID, id storage.FullReplicaID,
 ) error {
-	sl := stateloader.Make(rangeID)
+	sl := stateloader.Make(id.RangeID)
 	// Before creating the replica, see if there is a tombstone which would
 	// indicate that this replica has been removed.
 	// TODO(pav-kv): should also check that there is no existing replica, i.e.
 	// ReplicaID load should find nothing.
 	if ts, err := sl.LoadRangeTombstone(ctx, eng); err != nil {
 		return err
-	} else if replicaID < ts.NextReplicaID {
+	} else if id.ReplicaID < ts.NextReplicaID {
 		return &kvpb.RaftGroupDeletedError{}
 	}
 
@@ -140,12 +140,12 @@ func CreateUninitializedReplica(
 	// non-existent. The only RangeID-specific key that can be present is the
 	// RangeTombstone inspected above.
 	_ = CreateUninitReplicaTODO
-	if err := sl.SetRaftReplicaID(ctx, eng, replicaID); err != nil {
+	if err := sl.SetRaftReplicaID(ctx, eng, id.ReplicaID); err != nil {
 		return err
 	}
 
 	// Make sure that storage invariants for this uninitialized replica hold.
-	uninitDesc := roachpb.RangeDescriptor{RangeID: rangeID}
-	_, err := LoadReplicaState(ctx, eng, storeID, &uninitDesc, replicaID)
+	uninitDesc := roachpb.RangeDescriptor{RangeID: id.RangeID}
+	_, err := LoadReplicaState(ctx, eng, storeID, &uninitDesc, id.ReplicaID)
 	return err
 }

pkg/kv/kvserver/replica_init.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
@@ -79,7 +80,7 @@ func loadInitializedReplicaForTesting(
 func newInitializedReplica(
 	store *Store, loaded kvstorage.LoadedReplicaState, waitForPrevLeaseToExpire bool,
 ) (*Replica, error) {
-	r := newUninitializedReplicaWithoutRaftGroup(store, loaded.ReplState.Desc.RangeID, loaded.ReplicaID)
+	r := newUninitializedReplicaWithoutRaftGroup(store, loaded.FullReplicaID())
 	r.raftMu.Lock()
 	defer r.raftMu.Unlock()
 	r.mu.Lock()
@@ -99,10 +100,8 @@ func newInitializedReplica(
 //
 // TODO(#94912): we actually have another initialization path which should be
 // refactored: Replica.initFromSnapshotLockedRaftMuLocked().
-func newUninitializedReplica(
-	store *Store, rangeID roachpb.RangeID, replicaID roachpb.ReplicaID,
-) (*Replica, error) {
-	r := newUninitializedReplicaWithoutRaftGroup(store, rangeID, replicaID)
+func newUninitializedReplica(store *Store, id storage.FullReplicaID) (*Replica, error) {
+	r := newUninitializedReplicaWithoutRaftGroup(store, id)
 	r.raftMu.Lock()
 	defer r.raftMu.Unlock()
 	r.mu.Lock()
@@ -118,17 +117,15 @@ func newUninitializedReplica(
 // newUninitializedReplica() instead. This only exists for
 // newInitializedReplica() to avoid creating the Raft group twice (once when
 // creating the uninitialized replica, and once when initializing it).
-func newUninitializedReplicaWithoutRaftGroup(
-	store *Store, rangeID roachpb.RangeID, replicaID roachpb.ReplicaID,
-) *Replica {
-	uninitState := stateloader.UninitializedReplicaState(rangeID)
+func newUninitializedReplicaWithoutRaftGroup(store *Store, id storage.FullReplicaID) *Replica {
+	uninitState := stateloader.UninitializedReplicaState(id.RangeID)
 	r := &Replica{
 		AmbientContext: store.cfg.AmbientCtx,
-		RangeID:        rangeID,
-		replicaID:      replicaID,
+		RangeID:        id.RangeID,
+		replicaID:      id.ReplicaID,
 		creationTime:   timeutil.Now(),
 		store:          store,
-		abortSpan:      abortspan.New(rangeID),
+		abortSpan:      abortspan.New(id.RangeID),
 		concMgr: concurrency.NewManager(concurrency.Config{
 			NodeDesc:           store.nodeDesc,
 			RangeDesc:          uninitState.Desc,
@@ -145,7 +142,7 @@ func newUninitializedReplicaWithoutRaftGroup(
 		}),
 		allocatorToken: &plan.AllocatorToken{},
 	}
-	r.sideTransportClosedTimestamp.init(store.cfg.ClosedTimestampReceiver, rangeID)
+	r.sideTransportClosedTimestamp.init(store.cfg.ClosedTimestampReceiver, id.RangeID)
 	r.cachedClosedTimestampPolicy.Store(new(ctpb.RangeClosedTimestampPolicy))
 
 	r.mu.pendingLeaseRequest = makePendingLeaseRequest(r)
@@ -165,9 +162,9 @@ func newUninitializedReplicaWithoutRaftGroup(
 			// Expose proposal data for external test packages.
 			return store.cfg.TestingKnobs.TestingProposalSubmitFilter(kvserverbase.ProposalFilterArgs{
 				Ctx:        p.Context(),
-				RangeID:    rangeID,
+				RangeID:    id.RangeID,
 				StoreID:    store.StoreID(),
-				ReplicaID:  replicaID,
+				ReplicaID:  id.ReplicaID,
 				Cmd:        p.command,
 				QuotaAlloc: p.quotaAlloc,
 				CmdID:      p.idKey,
@@ -196,7 +193,7 @@ func newUninitializedReplicaWithoutRaftGroup(
 	// NB: state will be loaded when the replica gets initialized.
 	r.shMu.state = uninitState
 
-	r.rangeStr.store(replicaID, uninitState.Desc)
+	r.rangeStr.store(id.ReplicaID, uninitState.Desc)
 	// Add replica log tag - the value is rangeStr.String().
 	r.AmbientContext.AddLogTag("r", &r.rangeStr)
 	r.raftCtx = logtags.AddTag(r.AnnotateCtx(context.Background()), "raft", nil /* value */)
@@ -205,14 +202,14 @@ func newUninitializedReplicaWithoutRaftGroup(
 	// r.AmbientContext.AddLogTag("@", fmt.Sprintf("%x", unsafe.Pointer(r)))
 
 	r.raftMu.rangefeedCTLagObserver = newRangeFeedCTLagObserver()
-	r.raftMu.stateLoader = stateloader.Make(rangeID)
+	r.raftMu.stateLoader = stateloader.Make(id.RangeID)
 
 	// Initialize all the components of the log storage. The state of the log
 	// storage, such as RaftTruncatedState and the last entry ID, will be loaded
 	// when the replica is initialized.
 	sideloaded := logstore.NewDiskSideloadStorage(
 		store.cfg.Settings,
-		rangeID,
+		id.RangeID,
 		// NB: sideloaded log entries are persisted in the state engine so that they
 		// can be ingested to the state machine locally, when being applied.
 		store.StateEngine().GetAuxiliaryDir(),
@@ -229,13 +226,13 @@ func newUninitializedReplicaWithoutRaftGroup(
 	r.logStorage.mu.RWMutex = (*syncutil.RWMutex)(&r.mu.ReplicaMutex)
 	r.logStorage.raftMu.Mutex = &r.raftMu.Mutex
 	r.logStorage.ls = &logstore.LogStore{
-		RangeID:     rangeID,
+		RangeID:     id.RangeID,
 		Engine:      store.LogEngine(),
 		Sideload:    sideloaded,
 		StateLoader: r.raftMu.stateLoader.StateLoader,
 		// NOTE: use the same SyncWaiter loop for all raft log writes performed by a
 		// given range ID, to ensure that callbacks are processed in order.
-		SyncWaiter: store.syncWaiters[int(rangeID)%len(store.syncWaiters)],
+		SyncWaiter: store.syncWaiters[int(id.RangeID)%len(store.syncWaiters)],
 		Settings:   store.cfg.Settings,
 		DisableSyncLogWriteToss: buildutil.CrdbTestBuild &&
 			store.TestingKnobs().DisableSyncLogWriteToss,

pkg/kv/kvserver/replica_raft.go

@@ -2817,9 +2817,10 @@ func (r *Replica) acquireSplitLock(
 	ctx context.Context, split *roachpb.SplitTrigger,
 ) (func(), error) {
 	rightReplDesc, _ := split.RightDesc.GetReplicaDescriptor(r.StoreID())
-	rightRepl, _, err := r.store.getOrCreateReplica(
-		ctx, split.RightDesc.RangeID, rightReplDesc.ReplicaID, nil, /* creatingReplica */
-	)
+	rightRepl, _, err := r.store.getOrCreateReplica(ctx, storage.FullReplicaID{
+		RangeID:   split.RightDesc.RangeID,
+		ReplicaID: rightReplDesc.ReplicaID,
+	}, nil /* creatingReplica */)
 	// If getOrCreateReplica returns RaftGroupDeletedError we know that the RHS
 	// has already been removed. This case is handled properly in splitPostApply.
 	if errors.HasType(err, (*kvpb.RaftGroupDeletedError)(nil)) {
@@ -2852,9 +2853,10 @@ func (r *Replica) acquireMergeLock(
 	// complete, after which the replica will realize it has been destroyed and
 	// reject the snapshot.
 	rightReplDesc, _ := merge.RightDesc.GetReplicaDescriptor(r.StoreID())
-	rightRepl, _, err := r.store.getOrCreateReplica(
-		ctx, merge.RightDesc.RangeID, rightReplDesc.ReplicaID, nil, /* creatingReplica */
-	)
+	rightRepl, _, err := r.store.getOrCreateReplica(ctx, storage.FullReplicaID{
+		RangeID:   merge.RightDesc.RangeID,
+		ReplicaID: rightReplDesc.ReplicaID,
+	}, nil /* creatingReplica */)
 	if err != nil {
 		return nil, err
 	}

pkg/kv/kvserver/store_create_replica.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/errors"
@@ -51,13 +52,10 @@ var errRetry = errors.New("retry: orphaned replica")
 //
 // The caller must not hold the store's lock.
 func (s *Store) getOrCreateReplica(
-	ctx context.Context,
-	rangeID roachpb.RangeID,
-	replicaID roachpb.ReplicaID,
-	creatingReplica *roachpb.ReplicaDescriptor,
+	ctx context.Context, id storage.FullReplicaID, creatingReplica *roachpb.ReplicaDescriptor,
 ) (_ *Replica, created bool, _ error) {
-	if replicaID == 0 {
-		log.Fatalf(ctx, "cannot construct a Replica for range %d with 0 id", rangeID)
+	if id.ReplicaID == 0 {
+		log.Fatalf(ctx, "cannot construct a Replica for range %d with 0 id", id.RangeID)
 	}
 	// We need a retry loop as the replica we find in the map may be in the
 	// process of being removed or may need to be removed. Retries in the loop
@@ -71,12 +69,7 @@ func (s *Store) getOrCreateReplica(
 	})
 	for {
 		r.Next()
-		r, created, err := s.tryGetOrCreateReplica(
-			ctx,
-			rangeID,
-			replicaID,
-			creatingReplica,
-		)
+		r, created, err := s.tryGetOrCreateReplica(ctx, id, creatingReplica)
 		if errors.Is(err, errRetry) {
 			continue
 		}
@@ -92,12 +85,9 @@ func (s *Store) getOrCreateReplica(
 // removed. Returns errRetry error if the replica is in a transitional state and
 // its retrieval needs to be retried. Other errors are permanent.
 func (s *Store) tryGetReplica(
-	ctx context.Context,
-	rangeID roachpb.RangeID,
-	replicaID roachpb.ReplicaID,
-	creatingReplica *roachpb.ReplicaDescriptor,
+	ctx context.Context, id storage.FullReplicaID, creatingReplica *roachpb.ReplicaDescriptor,
 ) (*Replica, error) {
-	repl, found := s.mu.replicasByRangeID.Load(rangeID)
+	repl, found := s.mu.replicasByRangeID.Load(id.RangeID)
 	if !found {
 		return nil, nil
 	}
@@ -120,14 +110,14 @@ func (s *Store) tryGetReplica(
 	}
 
 	// The current replica needs to be removed, remove it and go back around.
-	if toTooOld := repl.replicaID < replicaID; toTooOld {
+	if toTooOld := repl.replicaID < id.ReplicaID; toTooOld {
 		if shouldLog := log.V(1); shouldLog {
 			log.Infof(ctx, "found message for replica ID %d which is newer than %v",
-				replicaID, repl)
+				id.ReplicaID, repl)
 		}
 
 		repl.mu.RUnlock()
-		if err := s.removeReplicaRaftMuLocked(ctx, repl, replicaID, "superseded by newer Replica", RemoveOptions{
+		if err := s.removeReplicaRaftMuLocked(ctx, repl, id.ReplicaID, "superseded by newer Replica", RemoveOptions{
 			DestroyData: true,
 		}); err != nil {
 			log.Fatalf(ctx, "failed to remove replica: %v", err)
@@ -137,17 +127,17 @@ func (s *Store) tryGetReplica(
 	}
 	defer repl.mu.RUnlock()
 
-	if repl.replicaID > replicaID {
+	if repl.replicaID > id.ReplicaID {
 		// The sender is behind and is sending to an old replica.
 		// We could silently drop this message but this way we'll inform the
 		// sender that they may no longer exist.
 		repl.raftMu.Unlock()
 		return nil, &kvpb.RaftGroupDeletedError{}
 	}
-	if repl.replicaID != replicaID {
+	if repl.replicaID != id.ReplicaID {
 		// This case should have been caught by handleToReplicaTooOld.
 		log.Fatalf(ctx, "intended replica id %d unexpectedly does not match the current replica %v",
-			replicaID, repl)
+			id.ReplicaID, repl)
 	}
 	return repl, nil
 }
@@ -159,13 +149,10 @@ func (s *Store) tryGetReplica(
 // tryGetOrCreateReplica will likely succeed, hence the loop in
 // getOrCreateReplica.
 func (s *Store) tryGetOrCreateReplica(
-	ctx context.Context,
-	rangeID roachpb.RangeID,
-	replicaID roachpb.ReplicaID,
-	creatingReplica *roachpb.ReplicaDescriptor,
+	ctx context.Context, id storage.FullReplicaID, creatingReplica *roachpb.ReplicaDescriptor,
 ) (_ *Replica, created bool, _ error) {
 	// The common case: look up an existing replica.
-	if repl, err := s.tryGetReplica(ctx, rangeID, replicaID, creatingReplica); err != nil {
+	if repl, err := s.tryGetReplica(ctx, id, creatingReplica); err != nil {
 		return nil, false, err
 	} else if repl != nil {
 		return repl, false, nil
@@ -175,24 +162,24 @@ func (s *Store) tryGetOrCreateReplica(
 	// be racing at this point, so grab a "lock" over this rangeID (represented by
 	// s.mu.creatingReplicas[rangeID]) by one goroutine, and retry others.
 	s.mu.Lock()
-	if _, ok := s.mu.creatingReplicas[rangeID]; ok {
+	if _, ok := s.mu.creatingReplicas[id.RangeID]; ok {
 		// Lost the race - another goroutine is currently creating that replica. Let
 		// the caller retry so that they can eventually see it.
 		s.mu.Unlock()
 		return nil, false, errRetry
 	}
-	s.mu.creatingReplicas[rangeID] = struct{}{}
+	s.mu.creatingReplicas[id.RangeID] = struct{}{}
 	s.mu.Unlock()
 	defer func() {
 		s.mu.Lock()
-		delete(s.mu.creatingReplicas, rangeID)
+		delete(s.mu.creatingReplicas, id.RangeID)
 		s.mu.Unlock()
 	}()
 	// Now we are the only goroutine trying to create a replica for this rangeID.
 
 	// Repeat the quick path in case someone has overtaken us while we were
 	// grabbing the "lock".
-	if repl, err := s.tryGetReplica(ctx, rangeID, replicaID, creatingReplica); err != nil {
+	if repl, err := s.tryGetReplica(ctx, id, creatingReplica); err != nil {
 		return nil, false, err
 	} else if repl != nil {
 		return repl, false, nil
@@ -204,16 +191,16 @@ func (s *Store) tryGetOrCreateReplica(
 	// Replica for this rangeID, and that's us.
 
 	_ = kvstorage.CreateUninitReplicaTODO
+	// TODO(sep-raft-log): needs both engines due to tombstone (which lives on
+	// statemachine).
 	if err := kvstorage.CreateUninitializedReplica(
-		// TODO(sep-raft-log): needs both engines due to tombstone (which lives on
-		// statemachine).
-		ctx, s.TODOEngine(), s.StoreID(), rangeID, replicaID,
+		ctx, s.TODOEngine(), s.StoreID(), id,
 	); err != nil {
 		return nil, false, err
 	}
 
 	// Create a new uninitialized replica and lock it for raft processing.
-	repl, err := newUninitializedReplica(s, rangeID, replicaID)
+	repl, err := newUninitializedReplica(s, id)
 	if err != nil {
 		return nil, false, err
 	}

pkg/kv/kvserver/store_raft.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/raft/raftpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
 	"github.com/cockroachdb/cockroach/pkg/util/grunning"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -421,12 +422,9 @@ func (s *Store) withReplicaForRequest(
 	f func(context.Context, *Replica) *kvpb.Error,
 ) *kvpb.Error {
 	// Lazily create the replica.
-	r, _, err := s.getOrCreateReplica(
-		ctx,
-		req.RangeID,
-		req.ToReplica.ReplicaID,
-		&req.FromReplica,
-	)
+	r, _, err := s.getOrCreateReplica(ctx, storage.FullReplicaID{
+		RangeID: req.RangeID, ReplicaID: req.ToReplica.ReplicaID,
+	}, &req.FromReplica)
 	if err != nil {
 		return kvpb.NewError(err)
 	}