kvserver: purge gc-able, unmigrated replicas during migrations

Fixes cockroachdb#58378.
Fixes cockroachdb#62267.

Previously it was possible for us to have replicas in-memory, with
pre-migrated state, even after a migration was finalized. This led to
the kind of badness we were observing in cockroachdb#62267, where it appeared that
a replica was not using the applied state key despite us having migrated
into it (see TruncatedAndRangeAppliedState, introduced in cockroachdb#58088).

---

To see how, consider the following set of events:

- Say r42 starts off on n1, n2, and n3
- n3 flaps and so we place a replica for r42 on n4
- n3's replica, r42/3, is now GC-able, but still un-GC-ed
- We run the applied state migration, first migrating all ranges into it
  and then purging outdated replicas
- Well, we should want to purge r42/3, cause it's un-migrated and
  evaluating anything on it (say a lease request) is unsound because
  we've bumped version gates that tell the kvserver to always expect
  post-migration state
- What happens when we try to purge r42/3? Previous to this PR if it
  didn't have a replica version, we'd skip over it (!)
- Was it possible for r42/3 to not have a replica version? Shouldn't it
  have been accounted for when we migrated all ranges? No, that's precisely
  why the migration infrastructure purge outdated replicas. The migrate
  request only returns once its applied on all followers; in our example
  that wouldn't include r42/3 since it was no longer one
- The stop-gap in cockroachdb#60429 made it so that we didn't GC r42/3, when we
  should've been doing the opposite. When iterating over a store's
  replicas for purging purposes, an empty replica version is fine and
  expected; we should interpret that as signal that we're dealing with a
  replica that was obviously never migrated (to even start using replica
  versions in the first place). Because it didn't have a valid replica
  version installed, we can infer that it's soon to be GC-ed (else we
  wouldn't have been able to finalize the applied state + replica
  version migration)
- The conditions above made it possible for us to evaluate requests on
  replicas with migration state out-of-date relative to the store's
  version
- Boom

Release note: None

pkg/kv/kvserver/client_migration_test.go

@@ -12,6 +12,7 @@ package kvserver_test
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"sync"
 	"testing"
@@ -41,69 +42,86 @@ import (
 func TestStorePurgeOutdatedReplicas(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	const numStores = 3
 
-	ctx := context.Background()
-	migrationVersion := roachpb.Version{Major: 42}
-	tc := testcluster.StartTestCluster(t, numStores,
-		base.TestClusterArgs{
-			ReplicationMode: base.ReplicationManual,
-			ServerArgs: base.TestServerArgs{
-				Knobs: base.TestingKnobs{
-					Store: &kvserver.StoreTestingKnobs{
-						DisableEagerReplicaRemoval: true,
-						DisableReplicaGCQueue:      true,
+	// The two sub-tests correspond to whether or not all replicas in the system
+	// come with a replica version installed. Replica versions were only
+	// introduced in the 21.1 cycle; replicas instantiated pre-21.1 have this
+	// unset. We'll want to test that PurgeOutdatedReplicas, when invoked, also
+	// clears out replicas without a version attached. These replicas
+	// are necessarily "outdated", else they'd be migrated into using replica
+	// versions (see clusterversion.ReplicaVersions).
+	for _, withInitialVersion := range []bool{true, false} {
+		t.Run(fmt.Sprintf("with-initial-version=%t", withInitialVersion), func(t *testing.T) {
+			const numStores = 3
+			ctx := context.Background()
+			migrationVersion := roachpb.Version{Major: 42}
+
+			storeKnobs := &kvserver.StoreTestingKnobs{
+				DisableEagerReplicaRemoval: true,
+				DisableReplicaGCQueue:      true,
+			}
+			if !withInitialVersion {
+				storeKnobs.InitialReplicaVersionOverride = &roachpb.Version{}
+			}
+
+			tc := testcluster.StartTestCluster(t, numStores,
+				base.TestClusterArgs{
+					ReplicationMode: base.ReplicationManual,
+					ServerArgs: base.TestServerArgs{
+						Knobs: base.TestingKnobs{
+							Store: storeKnobs,
+						},
 					},
 				},
-			},
-		},
-	)
-	defer tc.Stopper().Stop(context.Background())
-
-	// Create our scratch range and replicate it to n2 and n3.
-	n1, n2, n3 := 0, 1, 2
-	k := tc.ScratchRange(t)
-	tc.AddVotersOrFatal(t, k, tc.Target(n2), tc.Target(n3))
-	require.NoError(t, tc.WaitForVoters(k, tc.Target(n2), tc.Target(n3)))
-
-	for _, node := range []int{n2, n3} {
-		ts := tc.Servers[node]
-		store, pErr := ts.Stores().GetStore(ts.GetFirstStoreID())
-		if pErr != nil {
-			t.Fatal(pErr)
-		}
-
-		require.NotNil(t, store.LookupReplica(roachpb.RKey(k)))
-	}
-
-	// Mark the replica on n2 as eligible for GC.
-	desc := tc.RemoveVotersOrFatal(t, k, tc.Target(n2))
-
-	// We register an interceptor seeing as how we're attempting a (dummy) below
-	// raft migration below.
-	unregister := batcheval.TestingRegisterMigrationInterceptor(migrationVersion, func() {})
-	defer unregister()
-
-	// Migrate the remaining replicas on n1 and n3.
-	if err := tc.Server(n1).DB().Migrate(ctx, desc.StartKey, desc.EndKey, migrationVersion); err != nil {
-		t.Fatal(err)
-	}
-
-	ts := tc.Servers[n2]
-	store, pErr := ts.Stores().GetStore(ts.GetFirstStoreID())
-	if pErr != nil {
-		t.Fatal(pErr)
-	}
-
-	// Check to see that the replica still exists on n2.
-	require.NotNil(t, store.LookupReplica(roachpb.RKey(k)))
-
-	if err := store.PurgeOutdatedReplicas(ctx, migrationVersion); err != nil {
-		t.Fatal(err)
+			)
+			defer tc.Stopper().Stop(context.Background())
+
+			// Create our scratch range and replicate it to n2 and n3.
+			n1, n2, n3 := 0, 1, 2
+			k := tc.ScratchRange(t)
+			tc.AddVotersOrFatal(t, k, tc.Target(n2), tc.Target(n3))
+			require.NoError(t, tc.WaitForVoters(k, tc.Target(n2), tc.Target(n3)))
+
+			for _, node := range []int{n2, n3} {
+				ts := tc.Servers[node]
+				store, pErr := ts.Stores().GetStore(ts.GetFirstStoreID())
+				if pErr != nil {
+					t.Fatal(pErr)
+				}
+
+				require.NotNil(t, store.LookupReplica(roachpb.RKey(k)))
+			}
+
+			// Mark the replica on n2 as eligible for GC.
+			desc := tc.RemoveVotersOrFatal(t, k, tc.Target(n2))
+
+			// We register an interceptor seeing as how we're attempting a (dummy) below
+			// raft migration below.
+			unregister := batcheval.TestingRegisterMigrationInterceptor(migrationVersion, func() {})
+			defer unregister()
+
+			// Migrate the remaining replicas on n1 and n3.
+			if err := tc.Server(n1).DB().Migrate(ctx, desc.StartKey, desc.EndKey, migrationVersion); err != nil {
+				t.Fatal(err)
+			}
+
+			ts := tc.Servers[n2]
+			store, pErr := ts.Stores().GetStore(ts.GetFirstStoreID())
+			if pErr != nil {
+				t.Fatal(pErr)
+			}
+
+			// Check to see that the replica still exists on n2.
+			require.NotNil(t, store.LookupReplica(roachpb.RKey(k)))
+
+			if err := store.PurgeOutdatedReplicas(ctx, migrationVersion); err != nil {
+				t.Fatal(err)
+			}
+
+			// Check to see that the replica was purged from n2.
+			require.Nil(t, store.LookupReplica(roachpb.RKey(k)))
+		})
 	}
-
-	// Check to see that the replica was purged from n2.
-	require.Nil(t, store.LookupReplica(roachpb.RKey(k)))
 }
 
 // TestMigrateWithInflightSnapshot checks to see that the Migrate command blocks

pkg/kv/kvserver/replica.go

@@ -848,7 +848,22 @@ func (r *Replica) GetGCThreshold() hlc.Timestamp {
 // Version returns the replica version.
 func (r *Replica) Version() roachpb.Version {
 	if r.mu.state.Version == nil {
-		// TODO(irfansharif,tbg): This is a stop-gap for #58523.
+		// We introduced replica versions in v21.1 to service long-running
+		// migrations. For replicas that were instantiated pre-21.1, it's
+		// possible that the replica version is unset (but not for too long!).
+		//
+		// In the 21.1 cycle we introduced below-raft migrations that install a
+		// replica version on all replicas currently part of a raft group. What
+		// the migrations don't (directly) do is ensure that the versions are
+		// also installed on replicas slated to be GC-ed soon. For that purpose
+		// the migrations infrastructure makes use of PurgeOutdatedReplicas.
+		//
+		// All that is to say that in 21.1, it's possible we're dealing with
+		// unset replica versions.
+		//
+		// TODO(irfansharif): Remove this in 21.2; we'll have migrated into 21.1
+		// and purged all outdated replicas by then, and thus guaranteed to
+		// always have replica versions.
 		return roachpb.Version{}
 	}
 

pkg/kv/kvserver/store.go

@@ -2808,12 +2808,11 @@ func (s *Store) PurgeOutdatedReplicas(ctx context.Context, version roachpb.Versi
 	qp := quotapool.NewIntPool("purge-outdated-replicas", 50)
 	g := ctxgroup.WithContext(ctx)
 	s.VisitReplicas(func(repl *Replica) (wantMore bool) {
-		if (repl.Version() == roachpb.Version{}) {
-			// TODO(irfansharif,tbg): This is a stop gap for #58523.
-			return true
-		}
 		if !repl.Version().Less(version) {
-			// Nothing to do here.
+			// Nothing to do here. The less-than check also considers replicas
+			// with unset replica versions, which are only possible if they're
+			// left-over, GC-able replicas from before the first below-raft
+			// migration. We'll want to purge those.
 			return true
 		}
 

pkg/kv/kvserver/store_init.go

@@ -143,6 +143,10 @@ func WriteInitialClusterData(
 		return r
 	}
 
+	initialReplicaVersion := bootstrapVersion
+	if knobs.InitialReplicaVersionOverride != nil {
+		initialReplicaVersion = *knobs.InitialReplicaVersionOverride
+	}
 	// We iterate through the ranges backwards, since they all need to contribute
 	// to the stats of the first range (i.e. because they all write meta2 records
 	// in the first range), and so we want to create the first range last so that
@@ -236,12 +240,12 @@ func WriteInitialClusterData(
 
 		if tt := knobs.TruncatedStateTypeOverride; tt != nil {
 			if err := stateloader.WriteInitialRangeStateWithTruncatedState(
-				ctx, batch, *desc, bootstrapVersion, *tt,
+				ctx, batch, *desc, initialReplicaVersion, *tt,
 			); err != nil {
 				return err
 			}
 		} else {
-			if err := stateloader.WriteInitialRangeState(ctx, batch, *desc, bootstrapVersion); err != nil {
+			if err := stateloader.WriteInitialRangeState(ctx, batch, *desc, initialReplicaVersion); err != nil {
 				return err
 			}
 		}

pkg/kv/kvserver/testing_knobs.go

@@ -260,6 +260,10 @@ type StoreTestingKnobs struct {
 	// If set, use the given truncated state type when bootstrapping ranges.
 	// This is used for testing the truncated state migration.
 	TruncatedStateTypeOverride *stateloader.TruncatedStateType
+	// If set, use the given version as the initial replica version when
+	// bootstrapping ranges. This is used for testing the migration
+	// infrastructure.
+	InitialReplicaVersionOverride *roachpb.Version
 	// GossipWhenCapacityDeltaExceedsFraction specifies the fraction from the last
 	// gossiped store capacity values which need be exceeded before the store will
 	// gossip immediately without waiting for the periodic gossip interval.