kvnemesis: introduce safety and liveness modes

Currently, all kvnemesis test variants test both safety and liveness
properties: safety in the sense of validating serializability, and
liveness in the sense of ensuring availability.

With the introduction of various faults (e.g. network partitions),
kvnemesis tests can continue to operate the same way only if we
carefully craft the fault patterns in order to avoid unavailability.
While this is valuable, it is also important to test for safety testing
in the presense of unavailability.

This commit introduces two new modes of running kvnemesis tests, both of
which still validate serializability:
- Safety mode: all fault patterns are allowed; unavailability errors are
  ignored.
- Liveness mode: faults are introduced carefully to ensure a
  well-connected quorum is preserved; unavailability errors fail the
  test.

Fixes: #114814

Release note: None

Author: miraradeva

pkg/kv/kvnemesis/BUILD.bazel

@@ -96,8 +96,10 @@ go_test(
         "//pkg/security/securitytest",
         "//pkg/server",
         "//pkg/settings/cluster",
+        "//pkg/spanconfig",
         "//pkg/storage",
         "//pkg/storage/enginepb",
+        "//pkg/testutils",
         "//pkg/testutils/datapathutils",
         "//pkg/testutils/echotest",
         "//pkg/testutils/serverutils",

pkg/kv/kvnemesis/applier.go

@@ -7,6 +7,7 @@ package kvnemesis
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
@@ -98,7 +99,8 @@ func exceptUnhandledRetry(err error) bool {
 }
 
 func exceptAmbiguous(err error) bool { // true if ambiguous result
-	return errors.HasInterface(err, (*kvpb.ClientVisibleAmbiguousError)(nil))
+	return errors.HasInterface(err, (*kvpb.ClientVisibleAmbiguousError)(nil)) ||
+		strings.Contains(err.Error(), "result is ambiguous")
 }
 
 func exceptDelRangeUsingTombstoneStraddlesRangeBoundary(err error) bool {
@@ -109,6 +111,15 @@ func exceptConditionFailed(err error) bool {
 	return errors.HasType(err, (*kvpb.ConditionFailedError)(nil))
 }
 
+func exceptReplicaUnavailable(err error) bool {
+	return errors.HasType(err, (*kvpb.ReplicaUnavailableError)(nil))
+}
+
+func exceptContextCanceled(err error) bool {
+	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) ||
+		strings.Contains(err.Error(), "query execution canceled")
+}
+
 func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 	switch o := op.GetValue().(type) {
 	case *GetOperation,
@@ -751,7 +762,12 @@ func getRangeDesc(ctx context.Context, key roachpb.Key, dbs ...*kv.DB) roachpb.R
 	var opts = retry.Options{}
 	for r := retry.StartWithCtx(ctx, opts); r.Next(); dbIdx = (dbIdx + 1) % len(dbs) {
 		sender := dbs[dbIdx].NonTransactionalSender()
-		descs, _, err := kv.RangeLookup(ctx, sender, key, kvpb.CONSISTENT, 0, false)
+		// Use kvpb.INCONSISTENT because kv.CONSISTENT requires a transactional
+		// sender. In the generator, range lookups are usually used for finding
+		// replica/lease change targets, so it's ok if these are not consistent.
+		// Using kv.CONSISTENT with a non-transactional sender and in the presence
+		// of network partitions can lead to infinitely stuck lookups.
+		descs, _, err := kv.RangeLookup(ctx, sender, key, kvpb.INCONSISTENT, 0, false)
 		if err != nil {
 			log.Dev.Infof(ctx, "looking up descriptor for %s: %+v", key, err)
 			continue
@@ -762,12 +778,11 @@ func getRangeDesc(ctx context.Context, key roachpb.Key, dbs ...*kv.DB) roachpb.R
 		}
 		return descs[0]
 	}
-	panic(`unreachable`)
+	return roachpb.RangeDescriptor{}
 }
 
 func newGetReplicasFn(dbs ...*kv.DB) GetReplicasFn {
-	ctx := context.Background()
-	return func(key roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget) {
+	return func(ctx context.Context, key roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget) {
 		desc := getRangeDesc(ctx, key, dbs...)
 		replicas := desc.Replicas().Descriptors()
 		var voters []roachpb.ReplicationTarget

pkg/kv/kvnemesis/generator.go

@@ -15,6 +15,7 @@ import (
 	"slices"
 	"sort"
 	"sync/atomic"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvnemesis/kvnemesisutil"
@@ -28,6 +29,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"golang.org/x/exp/maps"
 )
@@ -645,7 +647,7 @@ func GeneratorDataSpan() roachpb.Span {
 
 // GetReplicasFn is a function that returns the current voting and non-voting
 // replicas, respectively, for the range containing a key.
-type GetReplicasFn func(roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget)
+type GetReplicasFn func(context.Context, roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget)
 
 // Generator incrementally constructs KV traffic designed to maximally test edge
 // cases.
@@ -676,7 +678,9 @@ type Generator struct {
 }
 
 // MakeGenerator constructs a Generator.
-func MakeGenerator(config GeneratorConfig, replicasFn GetReplicasFn) (*Generator, error) {
+func MakeGenerator(
+	config GeneratorConfig, replicasFn GetReplicasFn, mode TestMode,
+) (*Generator, error) {
 	if config.NumNodes <= 0 {
 		return nil, errors.Errorf(`NumNodes must be positive got: %d`, config.NumNodes)
 	}
@@ -693,7 +697,11 @@ func MakeGenerator(config GeneratorConfig, replicasFn GetReplicasFn) (*Generator
 	}
 	for i := 1; i <= config.NumNodes; i++ {
 		for j := 1; j <= config.NumNodes; j++ {
-			if i == j {
+			// In liveness mode, we don't allow adding partitions between the two
+			// protected nodes (node 1 and node 2), so we don't include those
+			// connections in the set of healthy connections at all.
+			protectedConn := (i == 1 && j == 2) || (i == 2 && j == 1)
+			if i == j || (mode == Liveness && protectedConn) {
 				continue
 			}
 			conn := connection{from: i, to: j}
@@ -708,6 +716,7 @@ func MakeGenerator(config GeneratorConfig, replicasFn GetReplicasFn) (*Generator
 		currentSplits:    make(map[string]struct{}),
 		historicalSplits: make(map[string]struct{}),
 		partitions:       p,
+		mode:             mode,
 	}
 	return g, nil
 }
@@ -745,6 +754,10 @@ type generator struct {
 	// partitions contains the sets of healthy and partitioned connections
 	// between nodes.
 	partitions
+
+	// mode is the test mode (e.g. Liveness or Safety). The generator needs it in
+	// order to set a timeout for range lookups under safety mode.
+	mode TestMode
 }
 
 type connection struct {
@@ -777,26 +790,45 @@ func (g *generator) RandStep(rng *rand.Rand) Step {
 	}
 
 	key := randKey(rng)
-	voters, nonVoters := g.replicasFn(roachpb.Key(key))
+	var voters, nonVoters []roachpb.ReplicationTarget
+	if g.mode == Safety {
+		if err := timeutil.RunWithTimeout(context.Background(), "getting replicas", 3*time.Second,
+			func(ctx context.Context) error {
+				voters, nonVoters = g.replicasFn(ctx, roachpb.Key(key))
+				return nil
+			}); err != nil {
+			voters, nonVoters = []roachpb.ReplicationTarget{}, []roachpb.ReplicationTarget{}
+		}
+	} else {
+		voters, nonVoters = g.replicasFn(context.Background(), roachpb.Key(key))
+	}
 	numVoters, numNonVoters := len(voters), len(nonVoters)
 	numReplicas := numVoters + numNonVoters
 	if numReplicas < g.Config.NumNodes {
-		addVoterFn := makeAddReplicaFn(key, voters, false /* atomicSwap */, true /* voter */)
-		addOpGen(&allowed, addVoterFn, g.Config.Ops.ChangeReplicas.AddVotingReplica)
-		addNonVoterFn := makeAddReplicaFn(key, nonVoters, false /* atomicSwap */, false /* voter */)
-		addOpGen(&allowed, addNonVoterFn, g.Config.Ops.ChangeReplicas.AddNonVotingReplica)
+		if len(voters) > 0 {
+			addVoterFn := makeAddReplicaFn(key, voters, false /* atomicSwap */, true /* voter */)
+			addOpGen(&allowed, addVoterFn, g.Config.Ops.ChangeReplicas.AddVotingReplica)
+		}
+		if len(nonVoters) > 0 {
+			addNonVoterFn := makeAddReplicaFn(key, nonVoters, false /* atomicSwap */, false /* voter */)
+			addOpGen(&allowed, addNonVoterFn, g.Config.Ops.ChangeReplicas.AddNonVotingReplica)
+		}
 	}
 	if numReplicas == g.Config.NumReplicas && numReplicas < g.Config.NumNodes {
-		atomicSwapVoterFn := makeAddReplicaFn(key, voters, true /* atomicSwap */, true /* voter */)
-		addOpGen(&allowed, atomicSwapVoterFn, g.Config.Ops.ChangeReplicas.AtomicSwapVotingReplica)
+		if len(voters) > 0 {
+			atomicSwapVoterFn := makeAddReplicaFn(key, voters, true /* atomicSwap */, true /* voter */)
+			addOpGen(&allowed, atomicSwapVoterFn, g.Config.Ops.ChangeReplicas.AtomicSwapVotingReplica)
+		}
 		if numNonVoters > 0 {
 			atomicSwapNonVoterFn := makeAddReplicaFn(key, nonVoters, true /* atomicSwap */, false /* voter */)
 			addOpGen(&allowed, atomicSwapNonVoterFn, g.Config.Ops.ChangeReplicas.AtomicSwapNonVotingReplica)
 		}
 	}
 	if numReplicas > g.Config.NumReplicas {
-		removeVoterFn := makeRemoveReplicaFn(key, voters, true /* voter */)
-		addOpGen(&allowed, removeVoterFn, g.Config.Ops.ChangeReplicas.RemoveVotingReplica)
+		if len(voters) > 0 {
+			removeVoterFn := makeRemoveReplicaFn(key, voters, true /* voter */)
+			addOpGen(&allowed, removeVoterFn, g.Config.Ops.ChangeReplicas.RemoveVotingReplica)
+		}
 		if numNonVoters > 0 {
 			removeNonVoterFn := makeRemoveReplicaFn(key, nonVoters, false /* voter */)
 			addOpGen(&allowed, removeNonVoterFn, g.Config.Ops.ChangeReplicas.RemoveNonVotingReplica)
@@ -810,8 +842,10 @@ func (g *generator) RandStep(rng *rand.Rand) Step {
 		promoteNonVoterFn := makePromoteReplicaFn(key, nonVoters)
 		addOpGen(&allowed, promoteNonVoterFn, g.Config.Ops.ChangeReplicas.PromoteReplica)
 	}
-	transferLeaseFn := makeTransferLeaseFn(key, append(voters, nonVoters...))
-	addOpGen(&allowed, transferLeaseFn, g.Config.Ops.ChangeLease.TransferLease)
+	if numVoters > 0 {
+		transferLeaseFn := makeTransferLeaseFn(key, append(voters, nonVoters...))
+		addOpGen(&allowed, transferLeaseFn, g.Config.Ops.ChangeLease.TransferLease)
+	}
 
 	addOpGen(&allowed, setLeaseType, g.Config.Ops.ChangeSetting.SetLeaseType)
 	addOpGen(&allowed, toggleGlobalReads, g.Config.Ops.ChangeZone.ToggleGlobalReads)

pkg/kv/kvnemesis/generator_test.go

@@ -73,11 +73,11 @@ func TestRandStep(t *testing.T) {
 	config := newAllOperationsConfig()
 	config.NumNodes, config.NumReplicas = 3, 2
 	rng, _ := randutil.NewTestRand()
-	getReplicasFn := func(_ roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget) {
+	getReplicasFn := func(ctx context.Context, _ roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget) {
 		return make([]roachpb.ReplicationTarget, rng.Intn(config.NumNodes)+1),
 			make([]roachpb.ReplicationTarget, rng.Intn(config.NumNodes)+1)
 	}
-	g, err := MakeGenerator(config, getReplicasFn)
+	g, err := MakeGenerator(config, getReplicasFn, 0)
 	require.NoError(t, err)
 
 	keys := make(map[string]struct{})

pkg/kv/kvnemesis/kvnemesis.go

@@ -15,11 +15,13 @@ import (
 	"reflect"
 	"strings"
 	"sync/atomic"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 )
 
@@ -66,6 +68,25 @@ func l(ctx context.Context, basename string, format string, args ...interface{})
 	return ""
 }
 
+// TestMode defines how faults are inserted and validated.
+type TestMode int
+
+const (
+	// The default value of TestMode is 0, which corresponds to no faults.
+	_ TestMode = iota
+	// Safety mode is used to test for safety properties (i.e. serializability) in
+	// the presence of unlimited faults. Unavailability errors are expected and
+	// ignored.
+	Safety = 1
+	// Liveness mode is used to test for liveness properties (i.e. availability).
+	// To do so in the presence of faults, the test will inject faults carefully,
+	// ensuring a well-connected quorum of replicas is always available, and the
+	// tests connects to one of the nodes in it. Without loss of generality, we
+	// keep nodes 1 and 2 available and connected to each other.
+	// TODO(mira): don't hardcode the protected nodes.
+	Liveness = 2
+)
+
 // RunNemesis generates and applies a series of Operations to exercise the KV
 // api. It returns a slice of the logical failures encountered.
 func RunNemesis(
@@ -75,6 +96,7 @@ func RunNemesis(
 	config GeneratorConfig,
 	concurrency int,
 	numSteps int,
+	mode TestMode,
 	dbs ...*kv.DB,
 ) ([]error, error) {
 	if env.L != nil {
@@ -86,11 +108,17 @@ func RunNemesis(
 
 	dataSpan := GeneratorDataSpan()
 
-	g, err := MakeGenerator(config, newGetReplicasFn(dbs...))
+	g, err := MakeGenerator(config, newGetReplicasFn(dbs...), mode)
 	if err != nil {
 		return nil, err
 	}
-	a := MakeApplier(env, dbs...)
+	applierDBs := dbs
+	// In Liveness mode, only nodes 1 and 2 are guaranteed to be available, so use
+	// only the first two DBs to apply operations.
+	if mode == Liveness && len(applierDBs) >= 2 {
+		applierDBs = applierDBs[:2]
+	}
+	a := MakeApplier(env, applierDBs...)
 	w, err := Watch(ctx, env, dbs, dataSpan)
 	if err != nil {
 		return nil, err
@@ -117,28 +145,35 @@ func RunNemesis(
 				step.format(&buf, formatCtx{indent: `  ` + workerName + ` PRE `})
 				l(ctx, basename, "%s", &buf)
 			}
-
-			trace, err := a.Apply(ctx, &step)
-			step.Trace = l(ctx, fmt.Sprintf("%s_trace", stepPrefix), "%s", trace.String())
-
-			stepsByWorker[workerIdx] = append(stepsByWorker[workerIdx], step)
-
-			prefix := ` OP  `
-			if err != nil {
-				prefix = ` ERR `
-			}
-
-			{
-				var buf strings.Builder
-				fmt.Fprintf(&buf, "  before: %s", step.Before)
-				step.format(&buf, formatCtx{indent: `  ` + workerName + prefix})
-				fmt.Fprintf(&buf, "\n  after: %s", step.After)
-				l(ctx, basename, "%s", &buf)
-			}
-
-			if err != nil {
+			applyAndLogOp := func(ctx context.Context) error {
+				trace, err := a.Apply(ctx, &step)
+				step.Trace = l(ctx, fmt.Sprintf("%s_trace", stepPrefix), "%s", trace.String())
+
+				stepsByWorker[workerIdx] = append(stepsByWorker[workerIdx], step)
+
+				prefix := ` OP  `
+				if err != nil {
+					prefix = ` ERR `
+				}
+
+				{
+					var buf strings.Builder
+					fmt.Fprintf(&buf, "  before: %s", step.Before)
+					step.format(&buf, formatCtx{indent: `  ` + workerName + prefix})
+					fmt.Fprintf(&buf, "\n  after: %s", step.After)
+					l(ctx, basename, "%s", &buf)
+				}
 				return err
 			}
+			if mode == Safety {
+				if err = timeutil.RunWithTimeout(ctx, "applying op", 10*time.Second, applyAndLogOp); err != nil {
+					return err
+				}
+			} else {
+				if err = applyAndLogOp(ctx); err != nil {
+					return err
+				}
+			}
 		}
 		return nil
 	}
@@ -162,12 +197,16 @@ func RunNemesis(
 	defer kvs.Close()
 
 	failures := Validate(allSteps, kvs, env.Tracker)
-
 	var filteredFailures []error
 	for _, f := range failures {
 		// ConditionFailedErrors are expected and can be ignored.
 		canBeIgnored := exceptConditionFailed(f)
-		if !canBeIgnored {
+		// The following errors are expected in safety mode.
+		canBeIgnoredSafety := mode == Safety &&
+			(exceptReplicaUnavailable(f) || exceptAmbiguous(f) || exceptContextCanceled(f))
+		// Ambiguous errors are expected in liveness mode.
+		canBeIgnoredLiveness := mode == Liveness && exceptAmbiguous(f)
+		if !canBeIgnored && !canBeIgnoredSafety && !canBeIgnoredLiveness {
 			filteredFailures = append(filteredFailures, f)
 		}
 	}