Merge #159853

159853: kvnemesis: enable node crashes to kvnemesis r=dodeca12 a=dodeca12

Previously, `kvnemesis` could only simulate graceful node stops and restarts, which limited its ability to test crash recovery scenarios. This was inadequate because real-world failures often involve abrupt crashes of nodes, leaving data in an inconsistent state that must be recovered on restart.

To address this, this PR adds crash operation support to `kvnemesis` through three commits:

1. Renames the `Restarter` interface to `ServerController` to better reflect its broader purpose of controlling server lifecycle operations, preparing for the addition of crash operations.

2. Extends `testcluster.TestCluster` with a `CrashServer` method that emulates a crash by stopping a server and creating a snapshot of its in-memory filesystems at the last sync point using `vfs.MemFS.CrashClone`. This simulates what would persist on disk after a real crash. The method also isolates the crashed node from peers by tripping circuit breakers, simulating network partition behavior. Adds `CrashNodeOperation` to the kvnemesis protobuf schema, integrates it into the generator to randomly crash nodes during test runs, implements crash application in the applier, and updates the validator to handle crash scenarios. The generator now tracks crashed nodes separately from stopped nodes.

3. Enables crash operations in kvnemesis test configurations by adding `removeRandCrashed` and `removeRandStoppedOrCrashed` methods to the nodes tracker. The `restartRandNode` function now randomly selects from both stopped and crashed nodes. Adds a new test `TestKVNemesisMultiNode_Crash_Liveness` that exercises crash operations with strict in-memory filesystem mode, which is required for proper crash simulation. 

Fixes #64828

Co-authored-by: Swapneeth Gorantla <[email protected]>

Author: craig[bot]

pkg/kv/kvnemesis/applier.go

@@ -182,14 +182,19 @@ func (a *Applier) applyOp(ctx context.Context, db *kv.DB, op *Operation) {
 		o.Result = resultInit(ctx, err)
 	case *StopNodeOperation:
 		serverID := int(o.NodeId) - 1
-		a.env.Restarter.StopServer(serverID)
+		a.env.ServerController.StopServer(serverID)
 		a.nodes.setStopped(int(o.NodeId))
 		o.Result = resultInit(ctx, nil)
 	case *RestartNodeOperation:
 		serverID := int(o.NodeId) - 1
-		err := a.env.Restarter.RestartServer(serverID)
+		err := a.env.ServerController.RestartServer(serverID)
 		a.nodes.setRunning(int(o.NodeId))
 		o.Result = resultInit(ctx, err)
+	case *CrashNodeOperation:
+		serverID := int(o.NodeId) - 1
+		a.env.ServerController.CrashNode(serverID)
+		a.nodes.setCrashed(int(o.NodeId))
+		o.Result = resultInit(ctx, nil)
 	case *ClosureTxnOperation:
 		// Use a backoff loop to avoid thrashing on txn aborts. Don't wait between
 		// epochs of the same transaction to avoid waiting while holding locks.

pkg/kv/kvnemesis/env.go

@@ -27,19 +27,20 @@ type Logger interface {
 	WriteFile(basename string, contents string) string
 }
 
-type Restarter interface {
+type ServerController interface {
 	StopServer(idx int)
 	RestartServer(idx int) error
+	CrashNode(idx int)
 }
 
 // Env manipulates the environment (cluster settings, zone configurations) that
 // the Applier operates in.
 type Env struct {
-	SQLDBs      []*gosql.DB
-	Tracker     *SeqTracker
-	L           Logger
-	Partitioner *rpc.Partitioner
-	Restarter   Restarter
+	SQLDBs           []*gosql.DB
+	Tracker          *SeqTracker
+	L                Logger
+	Partitioner      *rpc.Partitioner
+	ServerController ServerController
 }
 
 func (e *Env) anyNode() *gosql.DB {

pkg/kv/kvnemesis/generator.go

@@ -423,6 +423,8 @@ type FaultConfig struct {
 	StopNode int
 	// RestartNode is an operation that restarts a randomly chosen node.
 	RestartNode int
+	// CrashNode is an operation that crashes a randomly chosen node.
+	CrashNode int
 	// Disk stalls and other faults belong here.
 }
 
@@ -554,6 +556,7 @@ func newAllOperationsConfig() GeneratorConfig {
 			RemoveNetworkPartition: 1,
 			StopNode:               1,
 			RestartNode:            1,
+			CrashNode:              1,
 		},
 	}}
 }
@@ -654,6 +657,7 @@ func NewDefaultConfig() GeneratorConfig {
 	config.Ops.Fault.RemoveNetworkPartition = 0
 	config.Ops.Fault.StopNode = 0
 	config.Ops.Fault.RestartNode = 0
+	config.Ops.Fault.CrashNode = 0
 	return config
 }
 
@@ -811,6 +815,7 @@ type nodes struct {
 	mu      syncutil.RWMutex
 	running map[int]struct{}
 	stopped map[int]struct{}
+	crashed map[int]struct{}
 }
 
 func randNodeFromMap(m map[int]struct{}, rng *rand.Rand) int {
@@ -825,11 +830,26 @@ func (n *nodes) removeRandRunning(rng *rand.Rand) int {
 	return nodeID
 }
 
-func (n *nodes) removeRandStopped(rng *rand.Rand) int {
+// removeRandStoppedOrCrashed randomly picks a node from either the stopped or
+// crashed sets with uniform probability across all nodes.
+func (n *nodes) removeRandStoppedOrCrashed(rng *rand.Rand) int {
 	n.mu.Lock()
 	defer n.mu.Unlock()
-	nodeID := randNodeFromMap(n.stopped, rng)
-	delete(n.stopped, nodeID)
+	stopped := len(n.stopped)
+	crashed := len(n.crashed)
+
+	if stopped == 0 && crashed == 0 {
+		panic("no stopped or crashed nodes available")
+	}
+
+	var nodeID int
+	if rng.Intn(stopped+crashed) < stopped {
+		nodeID = randNodeFromMap(n.stopped, rng)
+		delete(n.stopped, nodeID)
+	} else {
+		nodeID = randNodeFromMap(n.crashed, rng)
+		delete(n.crashed, nodeID)
+	}
 	return nodeID
 }
 
@@ -845,6 +865,12 @@ func (n *nodes) setStopped(nodeID int) {
 	n.stopped[nodeID] = struct{}{}
 }
 
+func (n *nodes) setCrashed(nodeID int) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	n.crashed[nodeID] = struct{}{}
+}
+
 // RandStep returns a single randomly generated next operation to execute.
 //
 // RandStep is not concurrency safe.
@@ -929,9 +955,12 @@ func (g *generator) RandStep(rng *rand.Rand) Step {
 	if len(g.nodes.running) > 0 {
 		addOpGen(&allowed, stopRandNode, g.Config.Ops.Fault.StopNode)
 	}
-	if len(g.nodes.stopped) > 0 {
+	if len(g.nodes.stopped) > 0 || len(g.nodes.crashed) > 0 {
 		addOpGen(&allowed, restartRandNode, g.Config.Ops.Fault.RestartNode)
 	}
+	if len(g.nodes.running) > 0 {
+		addOpGen(&allowed, crashRandNode, g.Config.Ops.Fault.CrashNode)
+	}
 
 	return step(g.selectOp(rng, allowed))
 }
@@ -1877,10 +1906,15 @@ func stopRandNode(g *generator, rng *rand.Rand) Operation {
 }
 
 func restartRandNode(g *generator, rng *rand.Rand) Operation {
-	randNode := g.nodes.removeRandStopped(rng)
+	randNode := g.nodes.removeRandStoppedOrCrashed(rng)
 	return restartNode(randNode)
 }
 
+func crashRandNode(g *generator, rng *rand.Rand) Operation {
+	randNode := g.nodes.removeRandRunning(rng)
+	return crashNode(randNode)
+}
+
 func isFollowerReadEligibleOp(op Operation) bool {
 	if op.Get != nil && op.Get.FollowerReadEligible {
 		return true
@@ -2529,6 +2563,10 @@ func restartNode(nodeID int) Operation {
 	return Operation{RestartNode: &RestartNodeOperation{NodeId: int32(nodeID)}}
 }
 
+func crashNode(nodeID int) Operation {
+	return Operation{CrashNode: &CrashNodeOperation{NodeId: int32(nodeID)}}
+}
+
 type countingRandSource struct {
 	count atomic.Uint64
 	inner rand.Source64

pkg/kv/kvnemesis/generator_test.go

@@ -80,6 +80,7 @@ func TestRandStep(t *testing.T) {
 	n := nodes{
 		running: map[int]struct{}{1: {}, 2: {}, 3: {}},
 		stopped: make(map[int]struct{}),
+		crashed: make(map[int]struct{}),
 	}
 	g, err := MakeGenerator(config, getReplicasFn, 0, &n)
 	require.NoError(t, err)
@@ -444,6 +445,11 @@ func TestRandStep(t *testing.T) {
 			n.mu.Lock()
 			n.running[int(o.NodeId)] = struct{}{}
 			n.mu.Unlock()
+		case *CrashNodeOperation:
+			counts.Fault.CrashNode++
+			n.mu.Lock()
+			n.crashed[int(o.NodeId)] = struct{}{}
+			n.mu.Unlock()
 		default:
 			t.Fatalf("%T", o)
 		}

pkg/kv/kvnemesis/kvnemesis.go

@@ -109,6 +109,7 @@ func RunNemesis(
 	n := nodes{
 		running: make(map[int]struct{}),
 		stopped: make(map[int]struct{}),
+		crashed: make(map[int]struct{}),
 	}
 	for i := 1; i <= config.NumNodes; i++ {
 		// In liveness mode, we don't allow stopping and restarting the two
@@ -197,7 +198,7 @@ func RunNemesis(
 	}
 	env.Partitioner.EnablePartitions(false)
 	for i := 0; i < config.NumNodes; i++ {
-		_ = env.Restarter.RestartServer(i)
+		_ = env.ServerController.RestartServer(i)
 	}
 
 	allSteps := make(steps, 0, numSteps)

pkg/kv/kvnemesis/kvnemesis_test.go

@@ -270,7 +270,7 @@ func (cfg kvnemesisTestCfg) testClusterArgs(
 		}
 	}
 
-	reg := fs.NewStickyRegistry()
+	reg := fs.NewStickyRegistry(fs.UseStrictMemFS)
 	lisReg := listenerutil.NewListenerRegistry()
 	args := base.TestClusterArgs{
 		ReusableListenerReg: lisReg,
@@ -583,6 +583,47 @@ func TestKVNemesisMultiNode_Restart_Liveness(t *testing.T) {
 	})
 }
 
+func TestKVNemesisMultiNode_Crash(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testKVNemesisImpl(t, kvnemesisTestCfg{
+		numNodes:                     4,
+		numSteps:                     defaultNumSteps,
+		concurrency:                  5,
+		seedOverride:                 0,
+		invalidLeaseAppliedIndexProb: 0.2,
+		injectReproposalErrorProb:    0.2,
+		// assertRaftApply is disabled because the asserter assumes no node
+		// restarts (see asserter.go:29).
+		//
+		// Crashing nodes violates this assumption since node crashes differ
+		// from graceful restarts (abrupt stop vs a graceful shutdown).
+		//
+		// There's a race condition in crash simulation where
+		// CrashNode/CrashClone() can run while the server is still active, so
+		// WAL syncs and Raft applies can complete between the snapshot and
+		// stopServerLocked(). When assertRaftApply is true, the Asserter may
+		// record applies that aren't in the crash snapshot; after restart, the
+		// replica recovers from an earlier snapshot and re-applies, causing the
+		// Asserter to flag a false regression. This is a bug in the crash
+		// simulation code, not a bug in the database logic.
+		// TODO(dodeca12): resolve crash simulation bug for kvnemesis crashes for
+		// when assertRaftApply is true.
+		assertRaftApply: false,
+		mode:            Liveness,
+		testGeneratorConfig: func(cfg *GeneratorConfig) {
+			cfg.Ops.Fault.CrashNode = 1
+			cfg.Ops.Fault.RestartNode = 1
+
+			// Disallow replica changes because they interfere with the zone config
+			// constraints (at least one replica on nodes 1 and 2).
+			cfg.Ops.ChangeReplicas = ChangeReplicasConfig{}
+		},
+	},
+	)
+}
+
 func TestKVNemesisMultiNode(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -697,7 +738,7 @@ func testKVNemesisImpl(t testing.TB, cfg kvnemesisTestCfg) {
 
 	logger := newTBridge(t)
 	defer dumpRaftLogsOnFailure(t, logger.ll.dir, tc.Servers)
-	env := &Env{SQLDBs: sqlDBs, Tracker: tr, L: logger, Partitioner: &partitioner, Restarter: tc}
+	env := &Env{SQLDBs: sqlDBs, Tracker: tr, L: logger, Partitioner: &partitioner, ServerController: tc}
 	failures, err := RunNemesis(ctx, rng, env, config, cfg.concurrency, cfg.numSteps, cfg.mode, dbs...)
 
 	logMetricsReport(t, tc)

pkg/kv/kvnemesis/operations.go

@@ -74,6 +74,8 @@ func (op Operation) Result() *Result {
 		return &o.Result
 	case *RestartNodeOperation:
 		return &o.Result
+	case *CrashNodeOperation:
+		return &o.Result
 	default:
 		panic(errors.AssertionFailedf(`unknown operation: %T %v`, o, o))
 	}
@@ -254,6 +256,8 @@ func (op Operation) format(w *strings.Builder, fctx formatCtx) {
 		o.format(w, fctx)
 	case *RestartNodeOperation:
 		o.format(w, fctx)
+	case *CrashNodeOperation:
+		o.format(w, fctx)
 	default:
 		fmt.Fprintf(w, "%v", op.GetValue())
 	}
@@ -526,12 +530,17 @@ func (op RemoveNetworkPartitionOperation) format(w *strings.Builder, fctx format
 }
 
 func (op StopNodeOperation) format(w *strings.Builder, fctx formatCtx) {
-	fmt.Fprintf(w, `env.Restarter.StopNode(%d)`, int(op.NodeId))
+	fmt.Fprintf(w, `env.ServerController.StopNode(%d)`, int(op.NodeId))
 	op.Result.format(w)
 }
 
 func (op RestartNodeOperation) format(w *strings.Builder, fctx formatCtx) {
-	fmt.Fprintf(w, `env.Restarter.RestartNode(%d)`, int(op.NodeId))
+	fmt.Fprintf(w, `env.ServerController.RestartNode(%d)`, int(op.NodeId))
+	op.Result.format(w)
+}
+
+func (op CrashNodeOperation) format(w *strings.Builder, fctx formatCtx) {
+	fmt.Fprintf(w, `env.ServerController.CrashNode(%d)`, int(op.NodeId))
 	op.Result.format(w)
 }
 

pkg/kv/kvnemesis/operations.proto

@@ -206,6 +206,11 @@ message RestartNodeOperation {
   Result result = 2 [(gogoproto.nullable) = false];
 }
 
+message CrashNodeOperation {
+  int32 node_id = 1;
+  Result result = 2 [(gogoproto.nullable) = false];
+}
+
 message Operation {
   option (gogoproto.goproto_stringer) = false;
   option (gogoproto.onlyone) = true;
@@ -242,6 +247,7 @@ message Operation {
   RemoveNetworkPartitionOperation removeNetworkPartition = 28;
   StopNodeOperation stopNode = 29;
   RestartNodeOperation restartNode = 30;
+  CrashNodeOperation crashNode = 31;
 }
 
 enum ResultType {

pkg/kv/kvnemesis/validator.go

@@ -1024,6 +1024,9 @@ func (v *validator) processOp(op Operation) {
 	case *RestartNodeOperation:
 		execTimestampStrictlyOptional = true
 		v.checkError(op, t.Result)
+	case *CrashNodeOperation:
+		execTimestampStrictlyOptional = true
+		v.checkError(op, t.Result)
 	default:
 		panic(errors.AssertionFailedf(`unknown operation type: %T %v`, t, t))
 	}

pkg/testutils/testcluster/BUILD.bazel

@@ -48,6 +48,7 @@ go_library(
         "//pkg/util/tracing/tracingpb",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
+        "@com_github_cockroachdb_pebble//vfs",
         "@com_github_stretchr_testify//require",
     ],
 )