Don't return Ok(0) from poll_capacity

This still doesn't account for max send buffer size so this is not
perfect yet.

src/proto/streams/stream.rs

@@ -262,7 +262,6 @@ impl Stream {
 
     pub fn assign_capacity(&mut self, capacity: WindowSize) {
         debug_assert!(capacity > 0);
-        self.send_capacity_inc = true;
         self.send_flow.assign_capacity(capacity);
 
         tracing::trace!(
@@ -272,11 +271,7 @@ impl Stream {
             self.id
         );
 
-        // Only notify if the capacity exceeds the amount of buffered data
-        if self.send_flow.available() > self.buffered_send_data {
-            tracing::trace!("  notifying task");
-            self.notify_send();
-        }
+        self.notify_if_can_buffer_more();
     }
 
     /// If the capacity was limited because of the max_send_buffer_size,

tests/h2-support/src/util.rs

@@ -32,6 +32,7 @@ pub async fn yield_once() {
     .await;
 }
 
+/// Should only be called after a non-0 capacity was requested for the stream.
 pub fn wait_for_capacity(stream: h2::SendStream<Bytes>, target: usize) -> WaitForCapacity {
     WaitForCapacity {
         stream: Some(stream),
@@ -59,6 +60,11 @@ impl Future for WaitForCapacity {
 
             let act = self.stream().capacity();
 
+            // If a non-0 capacity was requested for the stream before calling
+            // wait_for_capacity, then poll_capacity should return Pending
+            // until there is a non-0 capacity.
+            assert_ne!(act, 0);
+
             if act >= self.target {
                 return Poll::Ready(self.stream.take().unwrap().into());
             }

tests/h2-tests/tests/flow_control.rs

@@ -1600,7 +1600,7 @@ async fn poll_capacity_after_send_data_and_reserve() {
         // Initial window size was 5 so current capacity is 0 even if we just reserved.
         assert_eq!(stream.capacity(), 0);
 
-        // The first call to `poll_capacity` in `wait_for_capacity` will return 0.
+        // This will panic if there is a bug causing h2 to return Ok(0) from poll_capacity.
         let mut stream = h2.drive(util::wait_for_capacity(stream, 5)).await;
 
         stream.send_data("".into(), true).unwrap();