kvnemesis: add crash operation support for fault injection testing

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 patch adds crash operation support to
`kvnemesis`. It 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. Circuit breakers automatically reset when the node restarts
and connections are re-established.

The patch 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.

Release note: None
Informs: #64828

Author: dodeca12

pkg/kv/kvnemesis/applier.go

@@ -190,6 +190,11 @@ func (a *Applier) applyOp(ctx context.Context, db *kv.DB, op *Operation) {
 		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

@@ -30,6 +30,7 @@ type Logger interface {
 type ServerController interface {
 	StopServer(idx int)
 	RestartServer(idx int) error
+	CrashNode(idx int)
 }
 
 // Env manipulates the environment (cluster settings, zone configurations) that

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

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)

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",
     ],
 )