Merge #29646 #29677

29646: roachtest: better merge testing in clearrange r=benesch a=tschottdorf

Unfortunately, the method to determine the range count is quite slow since
crdb_internal.ranges internally sends an RPC for each range to determine
the leaseholder.

Anecdotally, I've seen ~25% of the merges completed after less than 15
minutes. I know that it's slowing down over time, but @benesch will fix
that.

Also throws in aggressive consistency checks so that when something goes
out of sync, we find out right there.

Release note: None

29677: storage: preserve consistency when applying widening preemptive snapshots r=benesch a=benesch

Merges can cause preemptive snapshots that widen existing replicas. For
example, consider the following sequence of events:

1. A replica of range A is removed from store S, but is not garbage
   collected.
2. Range A subsumes its right neighbor B.
3. Range A is re-added to store S.

In step 3, S will receive a preemptive snapshot for A that requires
widening its existing replica, thanks to the intervening merge.

Problematically, the code to check whether this widening was possible,
in Store.canApplySnapshotLocked, was incorrectly mutating the range
descriptor in the snapshot header! Applying the snapshot would then fail
to clear all of the data from the old incarnation of the replica, since
the bounds on the range deletion tombstone were wrong. This often
resulted in replica inconsistency. Plus, the in-memory copy of the range
descriptor would be incorrect until the next descriptor update--though
this usually happened quickly, as the replica would apply the change
replicas command, which updates the descriptor, soon after applying the
preemptive snapshot.

To fix the problem, teach Store.canApplySnapshotLocked to make a copy of
the range descriptor before it mutates it.

To prevent regressions, add an assertion that a range's start key is
never changed to the descriptor update path. With this assertion in
place, but without the fix itself,
TestStoreRangeMergeReadoptedLHSFollower reliably fails.

Fixes #29252.

Release note: None

Co-authored-by: Tobias Schottdorf <[email protected]>
Co-authored-by: Nikhil Benesch <[email protected]>

Author: 3 people

pkg/cmd/roachtest/clearrange.go

@@ -23,6 +23,8 @@ import (
 )
 
 func registerClearRange(r *registry) {
+	const aggressiveConsistencyChecks = true
+
 	r.Add(testSpec{
 		Name:       `clearrange`,
 		MinVersion: `v2.1.0`,
@@ -52,7 +54,13 @@ func registerClearRange(r *registry) {
 			}
 
 			c.Put(ctx, cockroach, "./cockroach")
-			c.Start(ctx)
+			if aggressiveConsistencyChecks {
+				// Run with an env var that runs a synchronous consistency check after each rebalance and merge.
+				// This slows down merges, so it might hide some races.
+				c.Start(ctx, startArgs("--env=COCKROACH_CONSISTENCY_AGGRESSIVE=true"))
+			} else {
+				c.Start(ctx)
+			}
 
 			// Also restore a much smaller table. We'll use it to run queries against
 			// the cluster after having dropped the large table above, verifying that
@@ -68,6 +76,39 @@ func registerClearRange(r *registry) {
 
 			t.Status()
 
+			// Set up a convenience function that we can call to learn the number of
+			// ranges for the bank.bank table (even after it's been dropped).
+			numBankRanges := func() func() int {
+				conn := c.Conn(ctx, 1)
+				defer conn.Close()
+
+				var startHex string
+				// NB: set this to false to save yourself some time during development. Selecting
+				// from crdb_internal.ranges is very slow because it contacts all of the leaseholders.
+				// You may actually want to run a version of cockroach that doesn't do that because
+				// it'll still slow you down every time the method returned below is called.
+				if true {
+					if err := conn.QueryRow(
+						`SELECT to_hex(start_key) FROM crdb_internal.ranges WHERE "database" = 'bank' AND "table" = 'bank' ORDER BY start_key ASC LIMIT 1`,
+					).Scan(&startHex); err != nil {
+						t.Fatal(err)
+					}
+				} else {
+					startHex = "bd" // extremely likely to be the right thing (b'\275').
+				}
+				return func() int {
+					conn := c.Conn(ctx, 1)
+					defer conn.Close()
+					var n int
+					if err := conn.QueryRow(
+						`SELECT count(*) FROM crdb_internal.ranges WHERE substr(to_hex(start_key), 1, length($1::string)) = $1`, startHex,
+					).Scan(&n); err != nil {
+						t.Fatal(err)
+					}
+					return n
+				}
+			}()
+
 			m := newMonitor(ctx, c)
 			m.Go(func(ctx context.Context) error {
 				conn := c.Conn(ctx, 1)
@@ -77,6 +118,11 @@ func registerClearRange(r *registry) {
 					return err
 				}
 
+				// Merge as fast as possible to put maximum stress on the system.
+				if _, err := conn.ExecContext(ctx, `SET CLUSTER SETTING kv.range_merge.queue_interval = '0s'`); err != nil {
+					return err
+				}
+
 				t.WorkerStatus("dropping table")
 				defer t.WorkerStatus()
 
@@ -86,41 +132,46 @@ func registerClearRange(r *registry) {
 					return err
 				}
 
+				t.WorkerStatus("computing number of ranges")
+				initialBankRanges := numBankRanges()
+
+				t.WorkerStatus("dropping bank table")
 				if _, err := conn.ExecContext(ctx, `DROP TABLE bank.bank`); err != nil {
 					return err
 				}
 
-				// Spend a few minutes reading data with a timeout to make sure the
+				// Spend some time reading data with a timeout to make sure the
 				// DROP above didn't brick the cluster. At the time of writing,
-				// clearing all of the table data takes ~6min. We run for 2.5x that
-				// time to verify that nothing has gone wonky on the cluster.
-				//
-				// Don't lower this number, or the test may pass erroneously.
-				const minutes = 45
-				t.WorkerStatus("repeatedly running count(*) on small table")
-				for i := 0; i < minutes; i++ {
-					after := time.After(time.Minute)
+				// clearing all of the table data takes ~6min, so we want to run
+				// for at least a multiple of that duration.
+				const minDuration = 45 * time.Minute
+				deadline := timeutil.Now().Add(minDuration)
+				curBankRanges := numBankRanges()
+				t.WorkerStatus("waiting for ~", curBankRanges, " merges to complete (and for at least ", minDuration, " to pass)")
+				for timeutil.Now().Before(deadline) || curBankRanges > 1 {
+					after := time.After(5 * time.Minute)
+					curBankRanges = numBankRanges() // this call takes minutes, unfortunately
+					t.WorkerProgress(1 - float64(curBankRanges)/float64(initialBankRanges))
+
 					var count int
 					// NB: context cancellation in QueryRowContext does not work as expected.
 					// See #25435.
-					if _, err := conn.ExecContext(ctx, `SET statement_timeout = '10s'`); err != nil {
+					if _, err := conn.ExecContext(ctx, `SET statement_timeout = '5s'`); err != nil {
 						return err
 					}
-					// If we can't aggregate over 80kb in 10s, the database is far from usable.
-					start := timeutil.Now()
+					// If we can't aggregate over 80kb in 5s, the database is far from usable.
 					if err := conn.QueryRowContext(ctx, `SELECT count(*) FROM tinybank.bank`).Scan(&count); err != nil {
 						return err
 					}
-					c.l.Printf("read %d rows in %0.1fs\n", count, timeutil.Since(start).Seconds())
-					t.WorkerProgress(float64(i+1) / float64(minutes))
+
 					select {
 					case <-after:
 					case <-ctx.Done():
 						return ctx.Err()
 					}
 				}
-				// TODO(benesch): verify that every last range in the table has been
-				// merged away. For now, just exercising the merge code is a good start.
+				// TODO(tschottdorf): verify that disk space usage drops below to <some small amount>, but that
+				// may not actually happen (see https://github.com/cockroachdb/cockroach/issues/29290).
 				return nil
 			})
 			m.Wait()

pkg/storage/consistency_queue.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/gossip"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/grpcutil"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -33,6 +34,8 @@ var consistencyCheckInterval = settings.RegisterNonNegativeDurationSetting(
 	24*time.Hour,
 )
 
+var testingAggressiveConsistencyChecks = envutil.EnvOrDefaultBool("COCKROACH_CONSISTENCY_AGGRESSIVE", false)
+
 type consistencyQueue struct {
 	*baseQueue
 	interval       func() time.Duration

pkg/storage/merge_queue.go

@@ -273,6 +273,11 @@ func (mq *mergeQueue) process(
 		// as purgatory-worthy.
 		return rangeMergePurgatoryError{err}
 	}
+	if testingAggressiveConsistencyChecks {
+		if err := mq.store.consistencyQueue.process(ctx, lhsRepl, sysCfg); err != nil {
+			log.Warning(ctx, err)
+		}
+	}
 	return nil
 }
 

pkg/storage/replica.go

@@ -1717,6 +1717,11 @@ func (r *Replica) setDescWithoutProcessUpdate(desc *roachpb.RangeDescriptor) {
 		log.Fatalf(ctx, "cannot replace initialized descriptor with uninitialized one: %+v -> %+v",
 			r.mu.state.Desc, desc)
 	}
+	if r.mu.state.Desc != nil && !r.mu.state.Desc.StartKey.Equal(desc.StartKey) {
+		ctx := r.AnnotateCtx(context.TODO())
+		log.Fatalf(ctx, "attempted to change replica's start key from %s to %s",
+			r.mu.state.Desc.StartKey, desc.StartKey)
+	}
 
 	newMaxID := maxReplicaID(desc)
 	if newMaxID > r.mu.lastReplicaAdded {

pkg/storage/replica_raftstorage.go

@@ -945,6 +945,9 @@ func (r *Replica) applySnapshot(
 	// Update the range and store stats.
 	r.store.metrics.subtractMVCCStats(*r.mu.state.Stats)
 	r.store.metrics.addMVCCStats(*s.Stats)
+	// TODO(benesch): the next line updates r.mu.state.Desc, but that's supposed
+	// to be handled by the call to setDescWithoutProcessUpdate below. This is not
+	// a correctness issue right now, but it's liable to become one.
 	r.mu.state = s
 	r.assertStateLocked(ctx, r.store.Engine())
 	r.mu.Unlock()

pkg/storage/replicate_queue.go

@@ -265,6 +265,11 @@ func (rq *replicateQueue) process(
 			// Enqueue this replica again to see if there are more changes to be made.
 			rq.MaybeAdd(repl, rq.store.Clock().Now())
 		}
+		if testingAggressiveConsistencyChecks {
+			if err := rq.store.consistencyQueue.process(ctx, repl, sysCfg); err != nil {
+				log.Warning(ctx, err)
+			}
+		}
 		return nil
 	}
 	return errors.Errorf("failed to replicate after %d retries", retryOpts.MaxRetries)

pkg/storage/store_snapshot.go

@@ -394,7 +394,7 @@ func (s *Store) canApplySnapshot(
 func (s *Store) canApplySnapshotLocked(
 	ctx context.Context, snapHeader *SnapshotRequest_Header,
 ) (*ReplicaPlaceholder, error) {
-	desc := snapHeader.State.Desc
+	desc := *snapHeader.State.Desc
 	if v, ok := s.mu.replicas.Load(int64(desc.RangeID)); ok && (*Replica)(v).IsInitialized() {
 		// We have an initialized replica. Preemptive snapshots can be applied with
 		// no further checks if they do not widen the existing replica. Raft
@@ -420,7 +420,7 @@ func (s *Store) canApplySnapshotLocked(
 
 	// TODO(benesch): consider discovering and GC'ing *all* overlapping ranges,
 	// not just the first one that getOverlappingKeyRangeLocked happens to return.
-	if exRange := s.getOverlappingKeyRangeLocked(desc); exRange != nil {
+	if exRange := s.getOverlappingKeyRangeLocked(&desc); exRange != nil {
 		// We have a conflicting range, so we must block the snapshot.
 		// When such a conflict exists, it will be resolved by one range
 		// either being split or garbage collected.
@@ -454,7 +454,7 @@ func (s *Store) canApplySnapshotLocked(
 	}
 
 	placeholder := &ReplicaPlaceholder{
-		rangeDesc: *desc,
+		rangeDesc: desc,
 	}
 	return placeholder, nil
 }