Backport DD pipeline saturation test + gate-all-priorities (PR 13381) (#13411)

Ports apple/foundationdb#13381 to release-7.4:

- DDRelocationQueue: the relocation pipeline gate now holds moves of all
  priorities when the pipeline is full; only cancellations bypass (they
  reduce tracked metadata rather than add to it). Previously moves with
  priority >= PRIORITY_TEAM_UNHEALTHY passed through unconditionally. Add a
  "DD Pipeline Full" CODE_PROBE where pipelineFull is set.
- ServerKnobs: raise the buggified DD_MAX_PIPELINE_MOVES from 5 to 20 so
  simulation does not test under artificial scarcity (degenerate cases).
- Add fast/DDPipelineSaturation.toml, which saturates the back-pressure
  gate via many small shards plus failure-recovery relocations and trips
  the new code probe.

Paths adapted from main's layout (fdbserver/core, fdbserver/datadistributor)
to release-7.4 (fdbclient/ServerKnobs.cpp, fdbserver/DDRelocationQueue.actor.cpp);
the BUGGIFY macro is retained.

Testing: Release build; ran fdbserver -r simulation on DDPipelineSaturation
across several seeds, buggify on and off. The gate change is behaviorally
inert in these runs -- a baseline binary with the priority condition restored
produces bit-identical transaction counts and there are no SevError events --
and the "DD Pipeline Full" code probe is hit every run. The test currently
trips the Cycle workload's "Rate below desired rate" check because this branch
still carries the Cycle minimum-rate check that main removed in #13297; it
passes once that check is gone. Joshua validation deferred until the #13297
backport is sequenced onto the branch.

fdbclient/ServerKnobs.cpp

@@ -152,8 +152,10 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
 	// Hard cap on total relocations DD tracks (queued + in-flight). 1000 corresponds to a 500-server
 	// cluster with two concurrent shard moves per storage server.  We have observed large clusters doing
 	// 25-30GB in flight, or closer to 100 shards at a time, so this has plenty of margin of safety
-	// built in.
-	init( DD_MAX_PIPELINE_MOVES,                                1000 ); if( randomize && BUGGIFY ) DD_MAX_PIPELINE_MOVES = 5;
+	// built in.  For simulation, we don't really know how many servers there are, but 10 seems like a good
+	// guess (thus 20 moves).  Do not buggify this too small: testing under artificial scarcity results in
+	// uninteresting degenerate cases.
+	init( DD_MAX_PIPELINE_MOVES,                                1000 ); if( randomize && BUGGIFY ) DD_MAX_PIPELINE_MOVES = 20;
 	init( DD_REBALANCE_RESET_AMOUNT,                              30 );
 	init( INFLIGHT_PENALTY_HEALTHY,                              1.0 );
 	init( INFLIGHT_PENALTY_UNHEALTHY,                          500.0 );

fdbserver/DDRelocationQueue.actor.cpp

@@ -646,6 +646,7 @@ void DDQueue::updatePipelineFull() {
 		    .detail("PipelineSize", pipelineSize())
 		    .detail("PendingGateRelocations", pendingGateRelocations)
 		    .detail("PipelineLimit", SERVER_KNOBS->DD_MAX_PIPELINE_MOVES);
+		CODE_PROBE(true, "DD Pipeline Full");
 	} else if (pipelineSize() < SERVER_KNOBS->DD_MAX_PIPELINE_MOVES && pipelineFull->get()) {
 		pipelineFull->set(false);
 		TraceEvent("DDPipelineFullCleared", distributorId)
@@ -2834,10 +2835,10 @@ ACTOR Future<Void> BgDDLoadRebalance(DDQueue* self, int teamCollectionIndex, Dat
 	}
 }
 
-// Gates the relocation input stream by the pipeline limit. Cancellations and high-priority
-// moves (>= PRIORITY_TEAM_UNHEALTHY) always pass through immediately so that failure recovery
-// is never blocked by stuck or zombie moves holding pipeline slots. All other relocations are
-// held when the pipeline is full, waiting for pipelineFull to become false before forwarding.
+// Gates the relocation input stream by the pipeline limit. Cancellations always pass through
+// immediately because they reduce tracked metadata rather than adding to it. All other
+// relocations, regardless of priority, are held when the pipeline is full, waiting for
+// pipelineFull to become false before forwarding.
 // The global isDDPipelineControlEnabled() flag (cleared by disableDDPipelineControl()) also
 // bypasses the gate, allowing the test harness to open up the pipeline so DD can quiesce.
 // We poll it via delay() rather than AsyncVar to avoid cross-process callbacks in simulation.
@@ -2846,7 +2847,7 @@ ACTOR Future<Void> pipelineGateActor(Reference<DDQueue> self,
                                      PromiseStream<RelocateShard> output) {
 	loop {
 		state RelocateShard rs = waitNext(input);
-		if (!rs.cancelled && rs.priority < SERVER_KNOBS->PRIORITY_TEAM_UNHEALTHY) {
+		if (!rs.cancelled) {
 			while (self->pipelineFull->get() && isDDPipelineControlEnabled()) {
 				TraceEvent("DDPipelineFull", self->distributorId)
 				    .suppressFor(30.0)

tests/CMakeLists.txt

@@ -173,6 +173,7 @@ if(WITH_PYTHON)
   add_fdb_test(TEST_FILES rare/ChangeFeedOperations.toml)
   add_fdb_test(TEST_FILES rare/ChangeFeedOperationsMove.toml)
   add_fdb_test(TEST_FILES fast/DataLossRecovery.toml)
+  add_fdb_test(TEST_FILES fast/DDPipelineSaturation.toml)
   add_fdb_test(TEST_FILES fast/EncryptionOps.toml)
   # EncryptionUnitTests is only a convenience file to run the different types of encryption tests at once. Do not enable in general ensembles
   add_fdb_test(TEST_FILES fast/EncryptionUnitTests.toml IGNORE)

tests/fast/DDPipelineSaturation.toml

@@ -0,0 +1,35 @@
+# Exercises the Data Distribution relocation pipeline back-pressure gate
+# (DD_MAX_PIPELINE_MOVES). Forces many small shards and then a burst of
+# failure-recovery relocations (machine kills) so that pipelineSize()
+# exceeds the limit, tripping the "DD Pipeline Full" code probe in
+# DDRelocationQueue.actor.cpp. The limit is pinned to 20 -- the realistic
+# low end of the knob's range -- rather than an unrealistically small value
+# that would destabilize unrelated simulations.
+
+[[knobs]]
+dd_max_pipeline_moves = 20
+min_shard_bytes = 10000
+shard_bytes_per_sqrt_bytes = 0
+
+[[test]]
+testTitle = 'DDPipelineSaturation'
+
+    [[test.workload]]
+    testName = 'Cycle'
+    transactionsPerSecond = 5000.0
+    nodeCount = 60000
+    testDuration = 30.0
+
+    [[test.workload]]
+    testName = 'Attrition'
+    machinesToKill = 2
+    machinesToLeave = 5
+    reboot = true
+    testDuration = 30.0
+
+    [[test.workload]]
+    testName = 'Attrition'
+    machinesToKill = 2
+    machinesToLeave = 5
+    reboot = true
+    testDuration = 30.0