fix: auto data frame budget

src/client.rs

@@ -1364,9 +1364,9 @@ where
                 remote_reset_stream_max: builder.pending_accept_reset_stream_max,
                 local_error_reset_streams_max: builder.local_max_error_reset_streams,
                 settings: builder.settings,
-                data_frame_budget: builder
-                    .data_frame_budget
-                    .resolve(builder.initial_target_connection_window_size),
+                data_frame_budget: builder.data_frame_budget,
+                initial_target_connection_window_size: builder
+                    .initial_target_connection_window_size,
             },
         );
         let send_request = SendRequest {

src/proto/connection.rs

@@ -101,7 +101,11 @@ pub(crate) struct Config {
     pub remote_reset_stream_max: usize,
     pub local_error_reset_streams_max: Option<usize>,
     pub settings: frame::Settings,
-    pub data_frame_budget: usize,
+    /// How the connection-level DATA framing overhead budget is sized.
+    pub data_frame_budget: DataFrameBudget,
+    /// Target connection window an `Auto` budget is derived from. `None`
+    /// means the peer never configured one, so the protocol default applies.
+    pub initial_target_connection_window_size: Option<WindowSize>,
 }
 
 #[derive(Clone, Copy, Debug)]
@@ -163,6 +167,7 @@ where
                     .map(|max| max as usize),
                 local_max_error_reset_streams: config.local_error_reset_streams_max,
                 data_frame_budget: config.data_frame_budget,
+                initial_target_connection_window_size: config.initial_target_connection_window_size,
             }
         }
         let streams = Streams::new(streams_config(&config));
@@ -185,6 +190,14 @@ where
     }
 
     /// connection flow control
+    ///
+    /// PATCH(denoland): this also re-sizes an `Auto` DATA framing overhead
+    /// budget, which would otherwise stay pinned to the window the
+    /// connection was built with. Callers that autotune the connection
+    /// window at runtime -- hyper's `adaptive_window`, which hands h2 the
+    /// 64 KiB spec minimum up front and then grows the target from BDP
+    /// samples -- would otherwise run every connection on the smallest
+    /// budget `Auto` can resolve, no matter how large the window grows.
     pub(crate) fn set_target_window_size(&mut self, size: WindowSize) {
         let _res = self.inner.streams.set_target_connection_window_size(size);
         // TODO: proper error handling

src/proto/streams/counts.rs

@@ -25,6 +25,20 @@ impl Budget {
     fn replenish(&mut self, amount: usize) {
         self.available = self.available.saturating_add(amount).min(self.max);
     }
+
+    /// PATCH(denoland): raises the ceiling, crediting what was added to the
+    /// available balance.
+    ///
+    /// Grow-only on purpose: a peer that shrinks its window mid-connection
+    /// must not retroactively exhaust a budget an honest sender has already
+    /// spent against, which a plain reset would do.
+    fn grow_to(&mut self, max: usize) {
+        if max <= self.max {
+            return;
+        }
+        self.available = self.available.saturating_add(max - self.max);
+        self.max = max;
+    }
 }
 
 #[derive(Debug)]
@@ -70,6 +84,10 @@ pub(super) struct Counts {
     /// connection-level budget for DATA framing overhead.
     data_frame_budget: Budget,
 
+    /// PATCH(denoland): how `data_frame_budget` is sized, kept so an `Auto`
+    /// budget can be re-derived when the target connection window changes.
+    data_frame_budget_mode: DataFrameBudget,
+
     /// Number of empty, non-final DATA frames received over the lifetime of
     /// the connection.
     num_recv_empty_data_frames: usize,
@@ -90,11 +108,35 @@ impl Counts {
             num_remote_reset_streams: 0,
             max_local_error_reset_streams: config.local_max_error_reset_streams,
             num_local_error_reset_streams: 0,
-            data_frame_budget: Budget::new(config.data_frame_budget),
+            data_frame_budget: Budget::new(
+                config
+                    .data_frame_budget
+                    .resolve(config.initial_target_connection_window_size),
+            ),
+            data_frame_budget_mode: config.data_frame_budget,
             num_recv_empty_data_frames: 0,
         }
     }
 
+    /// PATCH(denoland): re-derives an `Auto` DATA framing overhead budget
+    /// from the new target connection window.
+    ///
+    /// `Auto` is documented as scaling with the connection window, but it is
+    /// resolved once from the window the connection was built with. A server
+    /// that autotunes the window afterwards -- hyper's `adaptive_window`
+    /// starts every connection at the 64 KiB spec minimum and grows the
+    /// target from BDP samples -- would otherwise be stuck on the smallest
+    /// budget `Auto` can produce for the life of the connection.
+    ///
+    /// A `Configured` budget is left alone: it is an explicit choice, not a
+    /// function of the window.
+    pub fn set_target_connection_window(&mut self, size: WindowSize) {
+        if let DataFrameBudget::Auto = self.data_frame_budget_mode {
+            self.data_frame_budget
+                .grow_to(DataFrameBudget::Auto.resolve(Some(size)));
+        }
+    }
+
     /// Records the framing overhead of a DATA frame.
     pub fn record_data_frame(&mut self, payload_len: usize) -> Result<(), BudgetExhausted> {
         if payload_len == 0 {
@@ -103,12 +145,30 @@ impl Counts {
                 .checked_add(1)
                 .ok_or(BudgetExhausted)?;
             if self.num_recv_empty_data_frames > MAX_RECV_EMPTY_DATA_FRAMES {
+                // PATCH(denoland): the two exhaustion paths share an error and
+                // a GOAWAY payload, so without distinct warn-level logs an
+                // operator cannot tell which rule killed a connection — the
+                // frame sizes only ever appear at debug level. One line per
+                // connection kill, so this cannot spam.
+                tracing::warn!(
+                    limit = MAX_RECV_EMPTY_DATA_FRAMES,
+                    "empty DATA frame limit exceeded; killing connection with ENHANCE_YOUR_CALM",
+                );
                 return Err(BudgetExhausted);
             }
             Ok(())
         } else if payload_len < DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD {
+            let budget = self.data_frame_budget.max;
             self.data_frame_budget
                 .consume(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - payload_len)
+                .inspect_err(|_| {
+                    // PATCH(denoland): see above.
+                    tracing::warn!(
+                        budget,
+                        payload_len,
+                        "small DATA frame budget exhausted; killing connection with ENHANCE_YOUR_CALM",
+                    );
+                })
         } else {
             self.data_frame_budget
                 .replenish(payload_len - DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD);
@@ -370,7 +430,8 @@ mod tests {
                 remote_init_window_sz: DEFAULT_INITIAL_WINDOW_SIZE,
                 remote_max_initiated: None,
                 local_max_error_reset_streams: None,
-                data_frame_budget: DEFAULT_DATA_FRAME_BUDGET,
+                data_frame_budget: DataFrameBudget::Auto,
+                initial_target_connection_window_size: None,
             },
         )
     }
@@ -384,6 +445,22 @@ mod tests {
         assert_eq!(budget.available, 10);
     }
 
+    #[test]
+    fn budget_grow_to_credits_the_difference() {
+        let mut budget = Budget::new(10);
+        budget.consume(10).unwrap();
+
+        budget.grow_to(30);
+        assert_eq!(budget.max, 30);
+        assert_eq!(budget.available, 20);
+
+        // Shrinking is a no-op, so an already-spent balance is never
+        // retroactively overdrawn.
+        budget.grow_to(5);
+        assert_eq!(budget.max, 30);
+        assert_eq!(budget.available, 20);
+    }
+
     #[test]
     fn budget_reports_exhaustion_without_underflowing() {
         let mut budget = Budget::new(10);
@@ -441,4 +518,58 @@ mod tests {
         }
         assert!(counts.record_data_frame(0).is_err());
     }
+
+    #[test]
+    fn auto_data_frame_budget_follows_target_connection_window() {
+        let mut counts = counts();
+        assert_eq!(
+            counts.data_frame_budget.max,
+            DEFAULT_INITIAL_WINDOW_SIZE as usize / 2
+        );
+
+        // Spend the starting budget one minimum-size frame at a time. This
+        // is the shape of a proxied upload whose chunks arrive smaller than
+        // the overhead threshold.
+        let per_frame = DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - 1;
+        let frames = counts.data_frame_budget.available / per_frame;
+        for _ in 0..frames {
+            counts.record_data_frame(1).unwrap();
+        }
+
+        // Growing the connection window has to grow the budget with it,
+        // otherwise a connection that started at the spec minimum stays on
+        // the smallest budget Auto can resolve however large the window
+        // gets, and the next small frame kills the whole connection.
+        counts.set_target_connection_window(1024 * 1024);
+        assert_eq!(counts.data_frame_budget.max, 512 * 1024);
+        for _ in 0..frames {
+            counts.record_data_frame(1).unwrap();
+        }
+    }
+
+    #[test]
+    fn auto_data_frame_budget_does_not_shrink_with_the_window() {
+        let mut counts = counts();
+        counts.set_target_connection_window(1024 * 1024);
+        counts.record_data_frame(1).unwrap();
+
+        counts.set_target_connection_window(DEFAULT_INITIAL_WINDOW_SIZE);
+
+        assert_eq!(counts.data_frame_budget.max, 512 * 1024);
+        assert_eq!(
+            counts.data_frame_budget.available,
+            512 * 1024 - (DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - 1)
+        );
+    }
+
+    #[test]
+    fn configured_data_frame_budget_ignores_target_connection_window() {
+        let mut counts = counts();
+        counts.data_frame_budget_mode = DataFrameBudget::Configured(1024);
+        counts.data_frame_budget = Budget::new(1024);
+
+        counts.set_target_connection_window(1024 * 1024);
+
+        assert_eq!(counts.data_frame_budget.max, 1024);
+    }
 }

src/proto/streams/mod.rs

@@ -79,10 +79,13 @@ pub struct Config {
     /// When this gets exceeded, we issue GOAWAYs.
     pub local_max_error_reset_streams: Option<usize>,
 
-    /// connection-level budget (in bytes) for DATA framing overhead.
-    ///
-    /// Default 25600 bytes
-    pub data_frame_budget: usize,
+    /// How the connection-level budget (in bytes) for DATA framing overhead
+    /// is sized. An `Auto` budget follows the target connection window; see
+    /// `Counts::set_target_connection_window`.
+    pub data_frame_budget: DataFrameBudget,
+
+    /// Target connection window the `Auto` budget is first derived from.
+    pub initial_target_connection_window_size: Option<WindowSize>,
 }
 
 trait DebugStructExt<'a, 'b> {

src/proto/streams/recv.rs

@@ -1318,7 +1318,8 @@ mod tests {
             remote_init_window_sz: DEFAULT_INITIAL_WINDOW_SIZE,
             remote_max_initiated: None,
             local_max_error_reset_streams: None,
-            data_frame_budget: DEFAULT_DATA_FRAME_BUDGET,
+            data_frame_budget: DataFrameBudget::Auto,
+            initial_target_connection_window_size: None,
         };
         let mut recv = Recv::new(peer::Dyn::Server, &config);
         let mut store = Store::new();

src/proto/streams/streams.rs

@@ -122,6 +122,10 @@ where
         let mut me = self.inner.lock().unwrap();
         let me = &mut *me;
 
+        // PATCH(denoland): keep an `Auto` DATA framing overhead budget in
+        // step with the window it is derived from.
+        me.counts.set_target_connection_window(size);
+
         me.actions
             .recv
             .set_target_connection_window(size, &mut me.actions.task)

src/server.rs

@@ -1535,10 +1535,10 @@ where
                                 .builder
                                 .local_max_error_reset_streams,
                             settings: self.builder.settings.clone(),
-                            data_frame_budget: self
+                            data_frame_budget: self.builder.data_frame_budget,
+                            initial_target_connection_window_size: self
                                 .builder
-                                .data_frame_budget
-                                .resolve(self.builder.initial_target_connection_window_size),
+                                .initial_target_connection_window_size,
                         },
                     );
 

tests/h2-tests/tests/server.rs

@@ -1922,9 +1922,8 @@ async fn graceful_shutdown_idle_connection() {
                 }
                 // GOAWAY: record the last-stream-id.
                 7 => {
-                    let last = u32::from_be_bytes([
-                        payload[0], payload[1], payload[2], payload[3],
-                    ]) & 0x7fff_ffff;
+                    let last = u32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]])
+                        & 0x7fff_ffff;
                     go_aways.push(last);
                 }
                 _ => {}