rac2: add kvflowsimulator equivalent simulation test

Add a new simulation test, `TestUsingSimulation`. This test mirrors
`kvflowsimulator.TestUsingSimulation` and its existing datadriven tests
cases.

When comparing the output from each, no behavioral differences show up,
only intentionally different metric names.

Note that like the `kvflowsimulator.TestUsingSimulation`, the simulation
test needs to be deterministic and therefore only uses a single
goroutine, to avoid goroutine scheduling causing non-determinism. As
such, helper methods have been added to facilitate waiting for available
tokens directly on a stream's token counter and on a range in a
non-blocking manner (`testingNonBlockingAdmit`). These added methods
are simplified versions of `WaitForEval`, which always wait for all
connected streams to have available tokens before admission.

Resolves: cockroachdb#130186
Release note: None

pkg/kv/kvserver/kvflowcontrol/rac2/BUILD.bazel

@@ -42,6 +42,7 @@ go_test(
         "log_tracker_test.go",
         "priority_test.go",
         "range_controller_test.go",
+        "simulation_test.go",
         "store_stream_test.go",
         "token_counter_test.go",
         "token_tracker_test.go",
@@ -61,17 +62,21 @@ go_test(
         "//pkg/testutils/datapathutils",
         "//pkg/testutils/echotest",
         "//pkg/util/admission/admissionpb",
+        "//pkg/util/asciitsdb",
         "//pkg/util/hlc",
         "//pkg/util/humanizeutil",
         "//pkg/util/leaktest",
         "//pkg/util/log",
+        "//pkg/util/metric",
         "//pkg/util/protoutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_dustin_go_humanize//:go-humanize",
         "@com_github_gogo_protobuf//jsonpb",
+        "@com_github_guptarohit_asciigraph//:asciigraph",
+        "@com_github_mkungla_bexp_v3//:bexp",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/kv/kvserver/kvflowcontrol/rac2/range_controller_test.go

@@ -232,6 +232,8 @@ func (s *testingRCState) getOrInitRange(t *testing.T, r testingRange) *testingRC
 		testRC = &testingRCRange{}
 		testRC.mu.r = r
 		testRC.mu.evals = make(map[string]*testingRCEval)
+		testRC.mu.outstandingReturns = make(map[roachpb.ReplicaID]kvflowcontrol.Tokens)
+		testRC.mu.quorumPosition = kvflowcontrolpb.RaftLogPosition{Term: 1, Index: 0}
 		options := RangeControllerOptions{
 			RangeID:             r.rangeID,
 			TenantID:            r.tenantID,
@@ -268,10 +270,23 @@ type testingRCEval struct {
 
 type testingRCRange struct {
 	rc *rangeController
+	// snapshots contain snapshots of the tracker state for different replicas,
+	// at various points in time. It is used in TestUsingSimulation.
+	snapshots []testingTrackerSnapshot
 
 	mu struct {
 		syncutil.Mutex
-		r     testingRange
+		r testingRange
+		// outstandingReturns is used in TestUsingSimulation to track token
+		// returns. Likewise, for quorumPosition. It is not used in
+		// TestRangeController.
+		outstandingReturns map[roachpb.ReplicaID]kvflowcontrol.Tokens
+		quorumPosition     kvflowcontrolpb.RaftLogPosition
+		// evals is used in TestRangeController for WaitForEval goroutine
+		// callbacks. It is not used in TestUsingSimulation, as the simulation test
+		// requires determinism on a smaller timescale than calling WaitForEval via
+		// multiple goroutines would allow. See testingNonBlockingAdmit to see how
+		// WaitForEval is done in simulation tests.
 		evals map[string]*testingRCEval
 	}
 }
@@ -329,6 +344,29 @@ type testingRange struct {
 	replicaSet     map[roachpb.ReplicaID]testingReplica
 }
 
+func makeSingleVoterTestingRange(
+	rangeID roachpb.RangeID,
+	tenantID roachpb.TenantID,
+	localNodeID roachpb.NodeID,
+	localStoreID roachpb.StoreID,
+) testingRange {
+	return testingRange{
+		rangeID:        rangeID,
+		tenantID:       tenantID,
+		localReplicaID: 1,
+		replicaSet: map[roachpb.ReplicaID]testingReplica{
+			1: {
+				desc: roachpb.ReplicaDescriptor{
+					NodeID:    localNodeID,
+					StoreID:   localStoreID,
+					ReplicaID: 1,
+					Type:      roachpb.VOTER_FULL,
+				},
+			},
+		},
+	}
+}
+
 func (t testingRange) replicas() ReplicaSet {
 	replicas := make(ReplicaSet, len(t.replicaSet))
 	for i, replica := range t.replicaSet {
@@ -588,12 +626,13 @@ func (t *testingProbeToCloseTimerScheduler) ScheduleSendStreamCloseRaftMuLocked(
 //
 //   - close_rcs: Closes all range controllers.
 //
-//   - admit: Admits the given store to the given range.
+//   - admit: Admits up to to_index for the given store, part of the given
+//     range.
 //     range_id=<range_id>
 //     store_id=<store_id> term=<term> to_index=<to_index> pri=<pri>
 //     ...
 //
-//   - raft_event: Simulates a raft event on the given rangeStateProbe, calling
+//   - raft_event: Simulates a raft event on the given range, calling
 //     HandleRaftEvent.
 //     range_id=<range_id>
 //     term=<term> index=<index> pri=<pri> size=<size>