diff --git a/src/client.rs b/src/client.rs index cb7e677a..9617d96c 100644 --- a/src/client.rs +++ b/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 { diff --git a/src/proto/connection.rs b/src/proto/connection.rs index dacd1d02..ecab2e5e 100644 --- a/src/proto/connection.rs +++ b/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, 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, } #[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 diff --git a/src/proto/streams/counts.rs b/src/proto/streams/counts.rs index 1cad3dfd..1eeb0c74 100644 --- a/src/proto/streams/counts.rs +++ b/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); + } } diff --git a/src/proto/streams/mod.rs b/src/proto/streams/mod.rs index c0845507..0ffa3a53 100644 --- a/src/proto/streams/mod.rs +++ b/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, - /// 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, } trait DebugStructExt<'a, 'b> { diff --git a/src/proto/streams/recv.rs b/src/proto/streams/recv.rs index c56b368f..4990b10b 100644 --- a/src/proto/streams/recv.rs +++ 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(); diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs index 1b73b333..f19f082a 100644 --- a/src/proto/streams/streams.rs +++ b/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) diff --git a/src/server.rs b/src/server.rs index 42a6dbae..3b31deb8 100644 --- a/src/server.rs +++ b/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, }, ); diff --git a/tests/h2-tests/tests/server.rs b/tests/h2-tests/tests/server.rs index b001d452..52858454 100644 --- a/tests/h2-tests/tests/server.rs +++ b/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); } _ => {}