Only toggle production state when above/below watermark (#2996) (#2997)

Motivation:

The high/low watermark backpressure strategy assumes that a yield can
never happen when there's no outstanding demand. It's reasonably easy
for this to not be a true: a handler decoding messages in a loop, for
example, may not pay attention to `read() calls and produce when there's
no outstanding demand. The channel should of course not read from the
socket, so produced values will stop being produced.

Modifications:

- Alter the high/low watermarks such that demand is only enabled when
the buffer depth is below the low watermark, and only disabled when
above the high watermark.

Result:

Fewer crashes

(cherry picked from commit ba6608e)

Sources/NIOCore/AsyncSequences/NIOAsyncSequenceProducerStrategies.swift

@@ -37,29 +37,20 @@ public enum NIOAsyncSequenceProducerBackPressureStrategies {
 
         public mutating func didYield(bufferDepth: Int) -> Bool {
             // We are demanding more until we reach the high watermark
-            if bufferDepth < self.highWatermark {
-                precondition(self.hasOustandingDemand)
-                return true
-            } else {
+            if bufferDepth >= self.highWatermark {
                 self.hasOustandingDemand = false
-                return false
             }
+
+            return self.hasOustandingDemand
         }
 
         public mutating func didConsume(bufferDepth: Int) -> Bool {
             // We start demanding again once we are below the low watermark
             if bufferDepth < self.lowWatermark {
-                if self.hasOustandingDemand {
-                    // We are below and have outstanding demand
-                    return true
-                } else {
-                    // We are below but don't have outstanding demand but need more
-                    self.hasOustandingDemand = true
-                    return true
-                }
-            } else {
-                return self.hasOustandingDemand
+                self.hasOustandingDemand = true
             }
+
+            return self.hasOustandingDemand
         }
     }
 }

Tests/NIOCoreTests/AsyncSequences/NIOAsyncSequenceProducer+HighLowWatermarkBackPressureStrategyTests.swift

@@ -57,4 +57,15 @@ final class NIOAsyncSequenceProducerBackPressureStrategiesHighLowWatermarkTests:
     func testDidConsume_whenAtLowWatermark() {
         XCTAssertTrue(self.strategy.didConsume(bufferDepth: 5))
     }
+
+    func testDidYieldWhenNoOutstandingDemand() {
+        // Hit the high watermark
+        XCTAssertFalse(self.strategy.didYield(bufferDepth: 10))
+        // Drop below it, don't read.
+        XCTAssertFalse(self.strategy.didConsume(bufferDepth: 7))
+        // Yield more, still above the low watermark, so don't produce more.
+        XCTAssertFalse(self.strategy.didYield(bufferDepth: 8))
+        // Drop below low watermark to start producing again.
+        XCTAssertTrue(self.strategy.didConsume(bufferDepth: 4))
+    }
 }