fix(rpc): Recover the congestion window after an overload backs it off

velox/exec/rpc/CongestionController.cpp

@@ -23,7 +23,8 @@ namespace facebook::velox::exec::rpc {
 
 void CongestionController::onError() {
   const auto prevEffective = effective_;
-  effective_ = std::max<int64_t>(effective_ / 2, minWindow_);
+  effective_ =
+      std::max<double>(effective_ / 2.0, static_cast<double>(minWindow_));
   if (effective_ < prevEffective) {
     ++numShrinks_;
   }
@@ -71,12 +72,15 @@ void CongestionController::onSample(int64_t rttNs) {
   // sqrt headroom keeps probing upward when latency is flat (gradient ~ 1), so
   // the window is never pinned by a fixed ceiling. stepCoef_ scales how hard it
   // probes (1.0 = the plain sqrt headroom).
-  const double headroom =
-      stepCoef_ * std::sqrt(static_cast<double>(effective_));
-  const auto newWindow = static_cast<int64_t>(
-      static_cast<double>(effective_) * gradient + headroom);
+  const double headroom = stepCoef_ * std::sqrt(effective_);
+  // Accumulate in floating point: one step is usually a fraction of a unit, and
+  // truncating here would discard it and pin the window.
+  const double newWindow = effective_ * gradient + headroom;
   const auto prevEffective = effective_;
-  effective_ = std::clamp(newWindow, minWindow_, maxWindow_);
+  effective_ = std::clamp<double>(
+      newWindow,
+      static_cast<double>(minWindow_),
+      static_cast<double>(maxWindow_));
   if (effective_ < prevEffective) {
     ++numShrinks_;
   }

velox/exec/rpc/CongestionController.h

@@ -38,6 +38,21 @@ namespace facebook::velox::exec::rpc {
 /// variance (e.g. variable LLM output length) because only queueing lifts the
 /// fastest request in a window.
 ///
+/// The window is carried as a double and reported as its floor. That is load
+/// bearing, not a style choice: one recomputation changes the window by
+/// stepCoef*sqrt(w) - w*(1 - gradient), which is well under 1 for most (w,
+/// gradient) pairs. Rounding that to an integer every window discards the
+/// increment, so an integer-valued window never grows except when the gradient
+/// is exactly 1. Accumulating in floating point lets the sub-unit steps add up,
+/// and the window tracks the law instead of freezing. While the baseline is
+/// held, the law has the fixed point
+///   w* = stepCoef^2 / (1 - gradient)^2,
+/// which is how to size stepCoef against a backend that needs at least N
+/// concurrent requests to reach its throughput knee: stepCoef >= sqrt(N) *
+/// (1 - gradient). That fixed point governs the transient only — under
+/// sustained elevation the baseline EMA absorbs the new latency, the gradient
+/// returns toward 1, and the window resumes probing upward by design.
+///
 /// A window constructed with startWindow == maxWindow that is never fed a
 /// sample stays fixed at that value — this is how callers pin a deterministic
 /// window (tests/config) without a separate code path.
@@ -81,12 +96,16 @@ class CongestionController {
         stepCoef_{std::max(0.0, stepCoef)},
         // Clamp the starting window into [minWindow_, maxWindow_] so limit()
         // is in range from construction, before the first onError/onSample.
-        effective_{std::clamp<int64_t>(startWindow, minWindow_, maxWindow_)} {}
+        effective_{std::clamp<double>(
+            static_cast<double>(startWindow),
+            static_cast<double>(minWindow_),
+            static_cast<double>(maxWindow_))} {}
 
   /// Returns the current admission limit (max in-flight units before
-  /// backpressure).
+  /// backpressure): the floor of the accumulated window, never below
+  /// minWindow.
   int64_t limit() const {
-    return effective_;
+    return static_cast<int64_t>(effective_);
   }
 
   /// Returns the learned baseline RTT (nanos), or 0 before the first full
@@ -121,8 +140,9 @@ class CongestionController {
   int64_t minWindow_{1};
   // Multiplier on the sqrt(window) additive-increase headroom.
   double stepCoef_{1.0};
-  // Current admission limit (the value limit() returns).
-  int64_t effective_{1};
+  // Accumulated admission window. Fractional so sub-unit growth is not lost
+  // between recomputations; limit() reports its floor.
+  double effective_{1.0};
 
   // Slow EMA of per-window minimum RTT (nanos); 0 until the first window.
   int64_t baselineRttNs_{0};

velox/exec/rpc/tests/CongestionControllerTest.cpp

@@ -249,5 +249,50 @@ TEST(CongestionControllerTest, baselineTracksSustainedLatencyAndRecovers) {
   EXPECT_GT(window.limit(), 4);
 }
 
+// Mild latency elevation must move the window off the floor. One
+// recomputation changes it by stepCoef*sqrt(w) - w*(1 - gradient); at w = 1
+// with gradient 0.8 that is +0.8, and an integer-valued window truncated 1.8
+// back to 1 and discarded it, every window, forever. Accumulating in floating
+// point lets those sub-unit steps add up. Measured over 200 windows of the
+// same input: 1 before this change, 1024 after.
+TEST(CongestionControllerTest, mildCongestionMovesWindowOffTheFloor) {
+  auto window = CongestionController{1, 1024};
+  feedWindow(window, 1'000'000); // baseline = 1ms
+  // A few windows at 1.25ms: gradient ~0.8, before the baseline has chased it.
+  for (int round = 0; round < 5; ++round) {
+    feedWindow(window, 1'250'000);
+  }
+  EXPECT_GT(window.limit(), 1);
+
+  // Sustained, the baseline absorbs the elevation and the window keeps
+  // probing; the point is that it climbs at all, not where it stops.
+  for (int round = 0; round < 195; ++round) {
+    feedWindow(window, 1'250'000);
+  }
+  EXPECT_GE(window.limit(), 16);
+}
+
+// A backend that batches internally gets *faster* as concurrency rises until
+// it reaches its knee. The window must climb through that region: latency
+// falling as the window grows reads as gradient 1, which is the grow signal.
+// The risk this pins down is the composite one -- a window parked below the
+// knee sees the backend's worst latency, which then looks like congestion.
+TEST(CongestionControllerTest, climbsThroughBatchingKnee) {
+  constexpr int64_t kKnee = 6;
+  auto window = CongestionController{1, 1024};
+  // Latency model: 10ms at a window of 1, falling to 1ms at the knee, then
+  // rising again with queueing beyond it.
+  auto rttFor = [](int64_t w) -> int64_t {
+    if (w <= kKnee) {
+      return 10'000'000 - (w - 1) * 1'500'000;
+    }
+    return 1'000'000 + (w - kKnee) * 500'000;
+  };
+  for (int round = 0; round < 100; ++round) {
+    feedWindow(window, rttFor(window.limit()));
+  }
+  EXPECT_GE(window.limit(), kKnee);
+}
+
 } // namespace
 } // namespace facebook::velox::exec::rpc