HTTP/2 on the client side (#797)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

apps/yaak-proxy/components/FlowDetail.tsx

@@ -69,8 +69,8 @@ export function FlowDetail({ style, sideBySide }: { style?: CSSProperties; sideB
 function FlowPane({ flow }: { flow: Flow }) {
   const [pretty, setPretty] = useState(true);
   const responseBody = useFlowBody(flow, "response", flow.responseTruncated);
-  // Failed before any response; an error after the headers is a body cut short mid-way.
-  const noResponse = flow.error != null && flow.status == null;
+  // Failed or cancelled before any response; after the headers, the body was cut short mid-way.
+  const noResponse = (flow.error != null || flow.cancelled) && flow.status == null;
   const mime = mimeEssence(findHeader(flow.responseHeaders, "content-type"));
   const queryCount = useMemo(() => queryParams(flow.url).length, [flow.url]);
   const openSave = useSetAtom(saveDialogFlowAtom);
@@ -133,11 +133,15 @@ function FlowPane({ flow }: { flow: Flow }) {
         )}
       >
         <HStack space={2} className="w-full shrink-0">
-          <HttpStatusTagRaw
-            showReason
-            status={flow.status ?? 0}
-            statusReason={flow.status == null ? null : statusReason(flow.status)}
-          />
+          {flow.cancelled && flow.status == null ? (
+            <span>Cancelled</span>
+          ) : (
+            <HttpStatusTagRaw
+              showReason
+              status={flow.status ?? 0}
+              statusReason={flow.status == null ? null : statusReason(flow.status)}
+            />
+          )}
           {flow.elapsedMs != null && (
             <>
               <span>&bull;</span>
@@ -197,6 +201,13 @@ function FlowPane({ flow }: { flow: Flow }) {
             {flow.error}
           </Banner>
         )}
+        {flow.cancelled && (
+          <Banner color="secondary" className="mx-3 mt-1 shrink-0 text-sm">
+            {flow.status == null
+              ? "The app cancelled the request before a response came back."
+              : "The app cancelled the request before the response finished."}
+          </Banner>
+        )}
         <Tabs
           tabs={tabs}
           label="Flow"
@@ -206,9 +217,14 @@ function FlowPane({ flow }: { flow: Flow }) {
           <TabContent value={TAB_RESPONSE}>
             {noResponse ? (
               <EmptyState>
-                <div className="text-text-subtle mb-1">No response</div>
+                <div className="text-text-subtle mb-1">
+                  {flow.cancelled ? "Cancelled" : "No response"}
+                </div>
                 <div className="text-xs">
-                  The request never got a response. The Request tab has what the app sent.
+                  {flow.cancelled
+                    ? "The app cancelled the request before a response came back."
+                    : "The request never got a response."}{" "}
+                  The Request tab has what the app sent.
                 </div>
               </EmptyState>
             ) : (

apps/yaak-proxy/components/FlowList.tsx

@@ -315,8 +315,8 @@ const FlowRow = memo(function FlowRow({
 }: RowProps) {
   const { host, path, query } = splitUrl(flow.url);
   const failed = flow.error != null;
-  // Failed before any response; an error after the headers is a body cut short mid-way.
-  const noResponse = failed && flow.status == null;
+  // Failed or cancelled before any response; after the headers, the body was cut short mid-way.
+  const noResponse = (failed || flow.cancelled) && flow.status == null;
   const type = shortContentType(flow.contentType);
 
   return (
@@ -342,6 +342,8 @@ const FlowRow = memo(function FlowRow({
           <Icon icon="bookmark" size="xs" color="primary" title="Saved to Yaak" />
         ) : failed ? (
           <Icon icon="circle_alert" size="xs" color="danger" />
+        ) : flow.cancelled ? (
+          <Icon icon="circle_off" size="xs" color="secondary" title="Cancelled by the app" />
         ) : flow.responseTruncated ? (
           <Icon icon="scissors" size="xs" color="notice" title="Body cut at the capture limit" />
         ) : null}
@@ -362,12 +364,18 @@ const FlowRow = memo(function FlowRow({
         {path}
         <span className="text-text-subtlest">{query}</span>
       </span>
-      <HttpStatusTagRaw status={flow.status ?? 0} short />
+      {flow.cancelled && flow.status == null ? (
+        <span className="font-mono text-text-subtlest">--</span>
+      ) : (
+        <HttpStatusTagRaw status={flow.status ?? 0} short />
+      )}
       {!sideBySide &&
         (failed ? (
           <span className="text-danger" title={flow.error ?? undefined}>
             {flow.error}
           </span>
+        ) : flow.cancelled ? (
+          <span className="text-text-subtle">Cancelled</span>
         ) : (
           <span className="text-text-subtle">
             {type ?? <span className="text-text-subtlest">--</span>}

apps/yaak-proxy/components/ProxyStatus.tsx

@@ -21,7 +21,7 @@ import { recordingLabel, systemProxySwitch } from "../lib/systemProxy";
 import { useShowToast } from "../lib/toast";
 
 /**
- * The header's state pill and its menu: recording, the system proxy and the certificate. The
+ * The header's state pill and its menu: recording, the system proxy, the certificate and HTTP/2. The
  * engine runs for as long as the app does, so the pill shows whether flows are being stored. A
  * certificate that is not trusted steps out of the menu beside the pill until it is.
  */
@@ -32,6 +32,7 @@ export function ProxyStatus() {
   const execute = useRpcMutation("execute_action");
   const retry = useRpcMutation("take_over");
   const setAtLaunch = useSetSystemProxyAtLaunch();
+  const setHttpVersion = useSetHttpVersion();
   const copyEnvironment = useCopyEnvironment();
   const { caStatus, trust, systemProxy, setSystemProxy } = setup;
   const systemSwitch = useSystemProxySwitch(setup);
@@ -168,6 +169,30 @@ export function ProxyStatus() {
         />
       ),
     },
+    {
+      type: "content",
+      label: (
+        <MenuRow
+          icon={<Icon icon="zap" size="sm" color="secondary" />}
+          title="HTTP/2"
+          description="Lets browsers send many requests at once. Turn off for apps that misbehave on it. Applies to new connections."
+          action={
+            <KeepMenuOpen>
+              <Checkbox
+                title="HTTP/2"
+                hideLabel
+                checked={proxyState?.httpVersion !== "http1Only"}
+                // The owner's engine would not see the change until it restarts.
+                disabled={other != null || setHttpVersion.isPending}
+                onChange={(enabled) =>
+                  setHttpVersion.mutate({ httpVersion: enabled ? "auto" : "http1Only" })
+                }
+              />
+            </KeepMenuOpen>
+          }
+        />
+      ),
+    },
     { type: "separator" },
     {
       label: "Copy Environment for a Terminal",
@@ -265,6 +290,14 @@ function useSetSystemProxyAtLaunch() {
   });
 }
 
+/** Saves the HTTP versions offered to clients, which the engine applies to new connections */
+function useSetHttpVersion() {
+  const store = useStore();
+  return useRpcMutation("set_http_version", {
+    onSettled: () => void refreshProxyState(store),
+  });
+}
+
 export function useCopyEnvironment() {
   const proxyState = useAtomValue(proxyStateAtom);
   const caStatus = useRpcQuery("ca_status", {});

apps/yaak-proxy/lib/flowActions.ts

@@ -54,7 +54,7 @@ export function useFlowActions(): (flow: Flow) => DropdownItem[] {
           showToast({ message: `Failed to copy ${label}: ${String(err)}`, color: "danger" });
         }
       };
-      const noResponse = flow.error != null && flow.status == null;
+      const noResponse = (flow.error != null || flow.cancelled) && flow.status == null;
       return [
         {
           label: flow.savedRequestId == null ? "Save as Yaak Request" : "Save to Yaak Again",

crates-proxy/yaak-proxy-lib/bindings/gen_models.ts

@@ -41,6 +41,11 @@ elapsedHeadersMs: number | null,
  * Milliseconds from `started_at` to the end of the response.
  */
 elapsedMs: number | null, error: string | null, 
+/**
+ * The client gave up before the response ended, by resetting its HTTP/2 stream or closing
+ * the connection. Not an error.
+ */
+cancelled: boolean, 
 /**
  * The id of the Yaak request this flow was last saved as.
  */

crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts

@@ -163,6 +163,10 @@ recording: boolean,
  * The system proxy turns on as the engine starts.
  */
 systemProxyAtLaunch: boolean, 
+/**
+ * The HTTP versions offered to clients.
+ */
+httpVersion: HttpVersion, 
 /**
  * Why the engine last failed to start, until it starts.
  */
@@ -175,6 +179,12 @@ otherOwner: OtherOwner | null, };
 
 export type GlobalAction = "recording_toggle" | "flows_clear";
 
+/**
+ * The HTTP versions offered to clients in intercepted HTTPS. Upstream, the engine speaks
+ * whichever the server picks either way.
+ */
+export type HttpVersion = "auto" | "http1Only";
+
 export type ListActionsRequest = Record<string, never>;
 
 export type ListActionsResponse = { actions: Array<[ActionInvocation, ActionMetadata]>, };
@@ -222,7 +232,7 @@ export type ProxyStatePayload = { state: ProxyState, };
 
 export type RpcEventSchema = { flows_cleared: FlowsClearedPayload, model_write: ModelPayload, proxy_state_changed: ProxyStatePayload, };
 
-export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], list_actions: [ListActionsRequest, ListActionsResponse], list_flows: [FlowFilter, ListFlowsResponse], set_recording: [SetRecordingRequest, boolean], set_system_proxy_at_launch: [SetSystemProxyAtLaunchRequest, boolean], app_icon: [AppIconRequest, string | null], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], clear_flows: [ClearFlowsRequest, number], export_har: [ExportHarRequest, number], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_workspaces: [ListWorkspacesRequest, ListWorkspacesResponse], preview_save: [PreviewSaveRequest, SavePreview], save_flow: [SaveFlowToYaakRequest, SavedFlow], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], take_over: [TakeOverRequest, boolean], };
+export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], list_actions: [ListActionsRequest, ListActionsResponse], list_flows: [FlowFilter, ListFlowsResponse], set_recording: [SetRecordingRequest, boolean], set_system_proxy_at_launch: [SetSystemProxyAtLaunchRequest, boolean], set_http_version: [SetHttpVersionRequest, HttpVersion], app_icon: [AppIconRequest, string | null], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], clear_flows: [ClearFlowsRequest, number], export_har: [ExportHarRequest, number], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_workspaces: [ListWorkspacesRequest, ListWorkspacesResponse], preview_save: [PreviewSaveRequest, SavePreview], save_flow: [SaveFlowToYaakRequest, SavedFlow], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], take_over: [TakeOverRequest, boolean], };
 
 export type SaveFlowToYaakRequest = { flowId: string, workspaceId: string, folderId?: string, 
 /**
@@ -272,6 +282,8 @@ export type ServiceProxy = {
  */
 name: string, http: ProxySetting, https: ProxySetting, };
 
+export type SetHttpVersionRequest = { httpVersion: HttpVersion, };
+
 export type SetRecordingRequest = { enabled: boolean, };
 
 export type SetSystemProxyAtLaunchRequest = { enabled: boolean, };

crates-proxy/yaak-proxy-lib/migrations/00000005_flow_cancelled.sql

@@ -0,0 +1,2 @@
+-- The client gave up on the flow, by resetting its HTTP/2 stream or closing the connection.
+ALTER TABLE flows ADD COLUMN cancelled BOOLEAN NOT NULL DEFAULT FALSE;

crates-proxy/yaak-proxy-lib/src/bodies.rs

@@ -37,8 +37,10 @@ pub async fn read_body(
 ) -> Result<DecodedBody, String> {
     let raw = read_raw_body(blobs, &flow.id, body)?;
     let stored_size = raw.len();
-    // A response that failed after its headers ended short on the wire.
-    let cut = body == FlowBody::Response && flow.status.is_some() && flow.error.is_some();
+    // A response that failed or was cancelled after its headers ended short on the wire.
+    let cut = body == FlowBody::Response
+        && flow.status.is_some()
+        && (flow.error.is_some() || flow.cancelled);
     let wire_truncated = flow.truncated(body) || cut;
     let encoding =
         ContentEncoding::from_header(find_header(flow.headers(body), "content-encoding"));
@@ -90,7 +92,39 @@ mod tests {
         let request = read_body(&blobs, &cut, FlowBody::Request, 1024).await.unwrap();
         assert!(!request.truncated);
         // With no headers there was never a response body to cut.
-        let refused = Flow { status: None, ..cut };
+        let refused = Flow { status: None, ..cut.clone() };
         assert!(!read_body(&blobs, &refused, FlowBody::Response, 1024).await.unwrap().truncated);
+
+        // The client cancelling partway through cuts it short the same way.
+        let cancelled = Flow { error: None, cancelled: true, ..cut };
+        let body = read_body(&blobs, &cancelled, FlowBody::Response, 1024).await.unwrap();
+        assert_eq!((body.data.as_slice(), body.truncated), (&b"{\"rows\": ["[..], true));
+    }
+
+    #[tokio::test]
+    async fn cancelled_gzip_response_keeps_what_decoded() {
+        use std::io::Write;
+        let (_db, blobs, _rx) = yaak_models::init_in_memory().unwrap();
+        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        gzip.write_all(&[b'z'; 4096]).unwrap();
+        let gzip = gzip.finish().unwrap();
+        let mut writer =
+            BodyChunkWriter::new(blobs.clone(), Flow::body_id("fl_gz", FlowBody::Response));
+        writer.write(&gzip[..gzip.len() / 2]).unwrap();
+        writer.finish().unwrap();
+        let flow = Flow {
+            id: "fl_gz".into(),
+            status: Some(200),
+            cancelled: true,
+            response_headers: vec![crate::models::ProxyHeader {
+                name: "content-encoding".into(),
+                value: "gzip".into(),
+            }],
+            ..Default::default()
+        };
+
+        let body = read_body(&blobs, &flow, FlowBody::Response, 8192).await.unwrap();
+        assert!(body.truncated);
+        assert_eq!(body.decode_error, None);
     }
 }

crates-proxy/yaak-proxy-lib/src/db.rs

@@ -140,6 +140,7 @@ mod tests {
             ALTER TABLE flows DROP COLUMN client_name;
             ALTER TABLE flows DROP COLUMN client_bundle_id;
             ALTER TABLE flows DROP COLUMN client_path;
+            ALTER TABLE flows DROP COLUMN cancelled;
             INSERT INTO flows (id, url, method) VALUES ('fl_old', 'https://example.com', 'GET');",
         )
         .unwrap();
@@ -150,6 +151,7 @@ mod tests {
         assert_eq!(flows.len(), 1);
         assert_eq!((flows[0].id.as_str(), flows[0].saved_request_id.as_deref()), ("fl_old", None));
         assert_eq!((flows[0].client_pid, flows[0].client_name.as_deref()), (None, None));
+        assert!(!flows[0].cancelled);
     }
 
     #[test]
@@ -194,6 +196,7 @@ mod tests {
             "client_name",
             "client_bundle_id",
             "client_path",
+            "cancelled",
         ] {
             assert!(flow_columns.contains(&column.to_string()), "{column} in {flow_columns:?}");
         }

crates-proxy/yaak-proxy-lib/src/lib.rs

@@ -23,7 +23,7 @@ use crate::db::ProxyQueryManager;
 use crate::macos::{SystemProxyBackend, SystemProxyStatus};
 use crate::models::{Flow, FlowBody, FlowIden, ModelPayload, ProxyHeader, find_header};
 use crate::save::{ClientDb, SaveFlowRequest, SaveFlowResponse, WorkspaceSummary};
-use crate::settings::ProxySettings;
+use crate::settings::{HttpVersion, ProxySettings};
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD as BASE64;
 use chrono::{DateTime, Utc};
@@ -169,6 +169,7 @@ impl ProxyCtx {
             .map_err(|e| format!("Failed to load proxy CA: {e}"))?;
         let mut config = ProxyConfig::new(port.unwrap_or(settings.port), Arc::new(ca));
         config.capture_body_limit = settings.capture_body_limit;
+        config.http2 = settings.http_version == HttpVersion::Auto;
         // Filled once the engine has bound, since the lookup matches sockets by its port.
         let attribution_slot: Arc<OnceLock<Attribution>> = Arc::new(OnceLock::new());
         let slot = attribution_slot.clone();
@@ -404,6 +405,20 @@ impl ProxyCtx {
         settings.save(&self.settings_path())
     }
 
+    /// The HTTP versions offered to clients, saved in `settings.json` and applied to the running
+    /// engine's next connections.
+    pub fn set_http_version(&self, version: HttpVersion) -> Result<(), String> {
+        let mut settings = self.settings()?;
+        settings.http_version = version;
+        settings.save(&self.settings_path())?;
+        if let Ok(handle) = self.handle.lock()
+            && let Some(handle) = handle.as_ref()
+        {
+            handle.set_http2(version == HttpVersion::Auto);
+        }
+        Ok(())
+    }
+
     /// Intercepts again the hosts passed through after clients rejected the CA, as they no
     /// longer will once it is trusted.
     pub fn forget_passthrough(&self) {
@@ -646,6 +661,8 @@ pub struct GetProxyStateResponse {
     pub recording: bool,
     /// The system proxy turns on as the engine starts.
     pub system_proxy_at_launch: bool,
+    /// The HTTP versions offered to clients.
+    pub http_version: HttpVersion,
     /// Why the engine last failed to start, until it starts.
     pub start_error: Option<String>,
     /// Another process that owns the engine, such as the CLI daemon. While one does, this app
@@ -680,6 +697,13 @@ pub struct SetSystemProxyAtLaunchRequest {
     pub enabled: bool,
 }
 
+#[derive(Deserialize, TS)]
+#[ts(export, export_to = "gen_rpc.ts")]
+#[serde(rename_all = "camelCase")]
+pub struct SetHttpVersionRequest {
+    pub http_version: HttpVersion,
+}
+
 #[derive(Deserialize, TS)]
 #[ts(export, export_to = "gen_rpc.ts")]
 pub struct AppIconRequest {
@@ -874,6 +898,7 @@ async fn get_proxy_state(
             configured_port: settings.port,
             recording: ctx.recording(),
             system_proxy_at_launch: settings.system_proxy_at_launch,
+            http_version: settings.http_version,
             start_error: ctx.start_error(),
             other_owner,
         })
@@ -894,6 +919,11 @@ fn set_system_proxy_at_launch(
     Ok(req.enabled)
 }
 
+fn set_http_version(ctx: &ProxyCtx, req: SetHttpVersionRequest) -> Result<HttpVersion, RpcError> {
+    ctx.set_http_version(req.http_version).map_err(rpc_error)?;
+    Ok(req.http_version)
+}
+
 async fn take_over(ctx: ProxyCtx, _req: TakeOverRequest) -> Result<bool, RpcError> {
     ctx.take_over().await.map_err(rpc_error)?;
     Ok(true)
@@ -1218,12 +1248,27 @@ fn run_event_loop(rx: std::sync::mpsc::Receiver<ProxyEvent>, sink: FlowSink) {
                     write_body_chunk(&mut f.response_body, &data);
                 }
             }
-            ProxyEvent::ResponseBodyComplete { id, size, truncated, elapsed_ms, error } => {
+            ProxyEvent::ResponseBodyComplete {
+                id,
+                size,
+                truncated,
+                elapsed_ms,
+                error,
+                cancelled,
+            } => {
                 if let Some(mut f) = in_flight.remove(&id) {
                     f.flow.response_size = size as i64;
                     f.flow.response_truncated = truncated;
                     f.flow.elapsed_ms = Some(clamp_ms(elapsed_ms));
                     f.flow.error = error;
+                    f.flow.cancelled = cancelled;
+                    finish_flow(&db, &events, &attribution, f);
+                }
+            }
+            ProxyEvent::Cancelled { id, elapsed_ms } => {
+                if let Some(mut f) = in_flight.remove(&id) {
+                    f.flow.elapsed_ms = Some(clamp_ms(elapsed_ms));
+                    f.flow.cancelled = true;
                     finish_flow(&db, &events, &attribution, f);
                 }
             }
@@ -1306,6 +1351,7 @@ define_rpc! {
         list_flows(flows::FlowFilter) -> ListFlowsResponse,
         set_recording(SetRecordingRequest) -> bool,
         set_system_proxy_at_launch(SetSystemProxyAtLaunchRequest) -> bool,
+        set_http_version(SetHttpVersionRequest) -> HttpVersion,
     }
     async_commands {
         app_icon(AppIconRequest) -> Option<String>,

crates-proxy/yaak-proxy-lib/src/models.rs

@@ -53,6 +53,9 @@ pub struct Flow {
     /// Milliseconds from `started_at` to the end of the response.
     pub elapsed_ms: Option<i32>,
     pub error: Option<String>,
+    /// The client gave up before the response ended, by resetting its HTTP/2 stream or closing
+    /// the connection. Not an error.
+    pub cancelled: bool,
     /// The id of the Yaak request this flow was last saved as.
     pub saved_request_id: Option<String>,
     /// The process that opened the client connection, where the platform can tell (macOS).
@@ -155,6 +158,7 @@ impl UpsertModelInfo for Flow {
             (ElapsedHeadersMs, self.elapsed_headers_ms.into()),
             (ElapsedMs, self.elapsed_ms.into()),
             (Error, self.error.into()),
+            (Cancelled, self.cancelled.into()),
             (SavedRequestId, self.saved_request_id.into()),
             (ClientPid, self.client_pid.into()),
             (ClientName, self.client_name.into()),
@@ -183,6 +187,7 @@ impl UpsertModelInfo for Flow {
             FlowIden::ElapsedHeadersMs,
             FlowIden::ElapsedMs,
             FlowIden::Error,
+            FlowIden::Cancelled,
             FlowIden::SavedRequestId,
             FlowIden::ClientPid,
             FlowIden::ClientName,
@@ -218,6 +223,7 @@ impl UpsertModelInfo for Flow {
             elapsed_headers_ms: r.get("elapsed_headers_ms")?,
             elapsed_ms: r.get("elapsed_ms")?,
             error: r.get("error")?,
+            cancelled: added_column(r, "cancelled")?.unwrap_or_default(),
             saved_request_id: added_column(r, "saved_request_id")?,
             client_pid: added_column(r, "client_pid")?,
             client_name: added_column(r, "client_name")?,

crates-proxy/yaak-proxy-lib/src/settings.rs

@@ -1,5 +1,6 @@
 use serde::{Deserialize, Serialize};
 use std::path::Path;
+use ts_rs::TS;
 use yaak_proxy::DEFAULT_CAPTURE_BODY_LIMIT;
 
 /// `settings.json` in the proxy data directory. Missing keys take their defaults.
@@ -14,6 +15,8 @@ pub struct ProxySettings {
     pub system_proxy_at_launch: bool,
     /// Bytes of each request and response body stored per flow.
     pub capture_body_limit: usize,
+    /// The HTTP versions offered to clients in intercepted HTTPS.
+    pub http_version: HttpVersion,
 }
 
 impl Default for ProxySettings {
@@ -22,10 +25,24 @@ impl Default for ProxySettings {
             port: 9090,
             system_proxy_at_launch: true,
             capture_body_limit: DEFAULT_CAPTURE_BODY_LIMIT,
+            http_version: HttpVersion::Auto,
         }
     }
 }
 
+/// The HTTP versions offered to clients in intercepted HTTPS. Upstream, the engine speaks
+/// whichever the server picks either way.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export, export_to = "gen_rpc.ts")]
+pub enum HttpVersion {
+    /// HTTP/2 or HTTP/1.1, as the client prefers.
+    #[default]
+    Auto,
+    /// HTTP/1.1 alone, for clients that misbehave on HTTP/2.
+    Http1Only,
+}
+
 impl ProxySettings {
     /// The settings at `path`, or the defaults when there is no file.
     pub fn load(path: &Path) -> Result<Self, String> {
@@ -70,8 +87,12 @@ mod tests {
     fn round_trip() {
         let dir = tempfile::tempdir().unwrap();
         let path = dir.path().join("settings.json");
-        let settings =
-            ProxySettings { port: 8080, system_proxy_at_launch: false, capture_body_limit: 1024 };
+        let settings = ProxySettings {
+            port: 8080,
+            system_proxy_at_launch: false,
+            capture_body_limit: 1024,
+            http_version: HttpVersion::Http1Only,
+        };
         settings.save(&path).unwrap();
         assert_eq!(ProxySettings::load(&path).unwrap(), settings);
 
@@ -79,10 +100,25 @@ mod tests {
             serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
         assert_eq!(
             json,
-            serde_json::json!({"port": 8080, "systemProxyAtLaunch": false, "captureBodyLimit": 1024})
+            serde_json::json!({
+                "port": 8080,
+                "systemProxyAtLaunch": false,
+                "captureBodyLimit": 1024,
+                "httpVersion": "http1Only",
+            })
         );
     }
 
+    #[test]
+    fn http_version_defaults_to_auto() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("settings.json");
+        std::fs::write(&path, r#"{"port": 8888}"#).unwrap();
+        assert_eq!(ProxySettings::load(&path).unwrap().http_version, HttpVersion::Auto);
+        std::fs::write(&path, r#"{"httpVersion": "auto"}"#).unwrap();
+        assert_eq!(ProxySettings::load(&path).unwrap().http_version, HttpVersion::Auto);
+    }
+
     #[test]
     fn the_earlier_system_proxy_key_carries_over() {
         let dir = tempfile::tempdir().unwrap();

crates-proxy/yaak-proxy-lib/tests/flow.rs

@@ -67,6 +67,7 @@ async fn route(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallib
             .header("content-encoding", "gzip")
             .body(Full::new(Bytes::new()))
             .unwrap(),
+        "/hang" => std::future::pending().await,
         // XML that only its body says is XML
         "/xml-as-text" => Response::builder()
             .header("content-type", "text/plain")
@@ -106,8 +107,12 @@ async fn wait_for_flow(ctx: &ProxyCtx, url: &str) -> Flow {
 #[tokio::test(flavor = "multi_thread")]
 async fn flow_is_stored_with_its_bodies() {
     let dir = tempfile::tempdir().unwrap();
-    let settings =
-        ProxySettings { port: 0, capture_body_limit: CAPTURE_LIMIT, system_proxy_at_launch: false };
+    let settings = ProxySettings {
+        port: 0,
+        capture_body_limit: CAPTURE_LIMIT,
+        system_proxy_at_launch: false,
+        ..Default::default()
+    };
     settings.save(&dir.path().join("settings.json")).unwrap();
 
     let (events, _event_rx) = RpcEventEmitter::new();
@@ -174,8 +179,12 @@ async fn flow_is_stored_with_its_bodies() {
 #[tokio::test(flavor = "multi_thread")]
 async fn early_response_keeps_request_totals() {
     let dir = tempfile::tempdir().unwrap();
-    let settings =
-        ProxySettings { port: 0, capture_body_limit: CAPTURE_LIMIT, system_proxy_at_launch: false };
+    let settings = ProxySettings {
+        port: 0,
+        capture_body_limit: CAPTURE_LIMIT,
+        system_proxy_at_launch: false,
+        ..Default::default()
+    };
     settings.save(&dir.path().join("settings.json")).unwrap();
 
     let (events, _event_rx) = RpcEventEmitter::new();
@@ -204,11 +213,84 @@ async fn rpc(ctx: &ProxyCtx, cmd: &str, payload: Value) -> Result<Value, String>
     yaak_proxy_lib::build_router().dispatch(cmd, payload, ctx).await.map_err(|e| e.message)
 }
 
+/// A request the client gives up on is stored as cancelled, not as an error.
+#[tokio::test(flavor = "multi_thread")]
+async fn cancelled_flow_is_not_an_error() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = ProxySettings { port: 0, system_proxy_at_launch: false, ..Default::default() };
+    settings.save(&dir.path().join("settings.json")).unwrap();
+
+    let (events, _event_rx) = RpcEventEmitter::new();
+    let ctx = ProxyCtx::new(dir.path(), events).unwrap();
+    ctx.start().unwrap();
+    let port = ctx.port().unwrap();
+
+    let upstream = spawn_upstream().await;
+    let client = reqwest::Client::builder()
+        .proxy(reqwest::Proxy::all(format!("http://127.0.0.1:{port}")).unwrap())
+        .timeout(Duration::from_millis(300))
+        .build()
+        .unwrap();
+    let url = format!("http://{upstream}/hang");
+    assert!(client.get(&url).send().await.unwrap_err().is_timeout());
+
+    let flow = wait_for_flow(&ctx, &url).await;
+    assert!(flow.cancelled);
+    assert_eq!((flow.error, flow.status), (None, None));
+    assert!(flow.elapsed_ms.is_some());
+}
+
+/// `http1Only` is saved and reaches the running engine: the next connection from a client that
+/// offers HTTP/2 speaks HTTP/1.1, and `auto` brings HTTP/2 back.
+#[tokio::test(flavor = "multi_thread")]
+async fn http_version_setting_applies_to_new_connections() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = ProxySettings { port: 0, system_proxy_at_launch: false, ..Default::default() };
+    settings.save(&dir.path().join("settings.json")).unwrap();
+
+    let (events, _event_rx) = RpcEventEmitter::new();
+    let ctx = ProxyCtx::new(dir.path(), events).unwrap();
+    ctx.start().unwrap();
+    let port = ctx.port().unwrap();
+    let ca = std::fs::read(dir.path().join("ca/ca.pem")).unwrap();
+    // Nothing listens there, so each flow is a 502, but its client side is spoken all the same.
+    let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap().port();
+
+    let mut sent = 0;
+    let mut client_version = async || {
+        sent += 1;
+        // A client of its own, so a connection of its own.
+        let client = reqwest::Client::builder()
+            .proxy(reqwest::Proxy::all(format!("http://127.0.0.1:{port}")).unwrap())
+            .add_root_certificate(reqwest::Certificate::from_pem(&ca).unwrap())
+            .build()
+            .unwrap();
+        let url = format!("https://localhost:{closed}/{sent}");
+        assert_eq!(client.get(&url).send().await.unwrap().status(), 502);
+        wait_for_flow(&ctx, &url).await.client_http_version
+    };
+
+    assert_eq!(client_version().await, "HTTP/2");
+    let set = rpc(&ctx, "set_http_version", json!({"httpVersion": "http1Only"})).await.unwrap();
+    assert_eq!(set, json!("http1Only"));
+    assert_eq!(rpc(&ctx, "get_proxy_state", json!({})).await.unwrap()["httpVersion"], "http1Only");
+    assert_eq!(client_version().await, "HTTP/1.1");
+
+    rpc(&ctx, "set_http_version", json!({"httpVersion": "auto"})).await.unwrap();
+    assert_eq!(client_version().await, "HTTP/2");
+    let saved = ProxySettings::load(&dir.path().join("settings.json")).unwrap();
+    assert_eq!(saved.http_version, yaak_proxy_lib::settings::HttpVersion::Auto);
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn body_rpcs() {
     let dir = tempfile::tempdir().unwrap();
-    let settings =
-        ProxySettings { port: 0, capture_body_limit: CAPTURE_LIMIT, system_proxy_at_launch: false };
+    let settings = ProxySettings {
+        port: 0,
+        capture_body_limit: CAPTURE_LIMIT,
+        system_proxy_at_launch: false,
+        ..Default::default()
+    };
     settings.save(&dir.path().join("settings.json")).unwrap();
 
     let (events, _event_rx) = RpcEventEmitter::new();
@@ -225,6 +307,7 @@ async fn body_rpcs() {
             "configuredPort": 0,
             "recording": true,
             "systemProxyAtLaunch": false,
+            "httpVersion": "auto",
             "startError": null,
             "otherOwner": null,
         })

crates/yaak-proxy/Cargo.toml

@@ -6,7 +6,7 @@ publish = false
 
 [dependencies]
 hyper = { version = "1", features = ["http1", "http2", "server", "client"] }
-hyper-util = { version = "0.1", features = ["tokio", "http1", "client-legacy"] }
+hyper-util = { version = "0.1", features = ["tokio", "http1", "client-legacy", "server-auto"] }
 http-body-util = "0.1"
 http = "1"
 bytes = "1"

crates/yaak-proxy/src/body.rs

@@ -1,5 +1,6 @@
 use std::pin::Pin;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::task::{Context, Poll};
 use std::time::Instant;
 
@@ -25,6 +26,8 @@ pub(crate) struct CaptureBody {
     captured: usize,
     size: u64,
     finished: bool,
+    /// Set when reading the body fails.
+    failed: Option<Arc<AtomicBool>>,
 }
 
 impl CaptureBody {
@@ -35,7 +38,23 @@ impl CaptureBody {
         start: Instant,
         engine: Arc<Engine>,
     ) -> Self {
-        Self { inner, direction, id, start, engine, captured: 0, size: 0, finished: false }
+        Self {
+            inner,
+            direction,
+            id,
+            start,
+            engine,
+            captured: 0,
+            size: 0,
+            finished: false,
+            failed: None,
+        }
+    }
+
+    /// Sets `failed` if reading the body fails, as a client's upload does when it cancels.
+    pub(crate) fn flag_failure(mut self, failed: Arc<AtomicBool>) -> Self {
+        self.failed = Some(failed);
+        self
     }
 
     /// Every byte the body declared has arrived. An HTTP/2 stream's end can come in a frame after
@@ -45,7 +64,7 @@ impl CaptureBody {
         self.inner.is_end_stream() || self.inner.size_hint().exact() == Some(0)
     }
 
-    fn finish(&mut self, error: Option<String>) {
+    fn finish(&mut self, error: Option<String>, cancelled: bool) {
         if self.finished {
             return;
         }
@@ -62,6 +81,7 @@ impl CaptureBody {
                 truncated,
                 elapsed_ms: self.start.elapsed().as_millis() as u64,
                 error,
+                cancelled,
             },
         };
         self.engine.emit(event);
@@ -92,14 +112,17 @@ impl Body for CaptureBody {
                     });
                 }
                 if self.complete() {
-                    self.finish(None);
+                    self.finish(None, false);
                 }
             }
             Poll::Ready(Some(Err(e))) => {
+                if let Some(failed) = &self.failed {
+                    failed.store(true, Ordering::Relaxed);
+                }
                 let error = e.to_string();
-                self.finish(Some(error));
+                self.finish(Some(error), false);
             }
-            Poll::Ready(None) => self.finish(None),
+            Poll::Ready(None) => self.finish(None, false),
             Poll::Pending => {}
         }
         result
@@ -114,10 +137,11 @@ impl Body for CaptureBody {
     }
 }
 
+/// A response body dropped before its end was abandoned by the client, which reset its stream or
+/// closed the connection. Dropping the upstream body with it cancels the request there too.
 impl Drop for CaptureBody {
     fn drop(&mut self) {
-        let error = (self.direction == Direction::Response && !self.complete())
-            .then(|| "Client closed the connection before the response finished".to_string());
-        self.finish(error);
+        let cancelled = self.direction == Direction::Response && !self.complete();
+        self.finish(None, cancelled);
     }
 }

crates/yaak-proxy/src/cert.rs

@@ -24,7 +24,8 @@ pub struct CaStore {
     cert_der: CertificateDer<'static>,
     issuer: Certificate,
     key: KeyPair,
-    leaf_configs: Mutex<HashMap<String, Arc<ServerConfig>>>,
+    /// Keyed by host and whether HTTP/2 is offered.
+    leaf_configs: Mutex<HashMap<(String, bool), Arc<ServerConfig>>>,
 }
 
 impl CaStore {
@@ -92,14 +93,50 @@ impl CaStore {
         &self.key_path
     }
 
+    /// The TLS config presented to clients of `host`, offering `h2` in ALPN when `http2` is set
+    /// and `http/1.1` always.
     pub(crate) fn server_config(
         &self,
         host: &str,
-    ) -> Result<Arc<ServerConfig>, Box<dyn std::error::Error + Send + Sync>> {
-        if let Some(config) = self.leaf_configs.lock().unwrap().get(host) {
-            return Ok(config.clone());
+        http2: bool,
+    ) -> Result<Arc<ServerConfig>, BoxError> {
+        self.server_config_with(host, http2, || self.sign_leaf(host))
+    }
+
+    /// [`Self::server_config`], with `sign` making a new leaf. It runs outside the cache's lock,
+    /// so a host being signed holds up no other host's lookup. Connections racing to the same
+    /// new host may each sign one; the first stored is the one kept.
+    fn server_config_with(
+        &self,
+        host: &str,
+        http2: bool,
+        sign: impl FnOnce() -> Result<ServerConfig, BoxError>,
+    ) -> Result<Arc<ServerConfig>, BoxError> {
+        let key = (host.to_string(), http2);
+        {
+            let mut configs = self.leaf_configs.lock().unwrap();
+            if let Some(config) = configs.get(&key) {
+                return Ok(config.clone());
+            }
+            // The same leaf serves both, so switching HTTP/2 off and on signs nothing new.
+            if let Some(other) = configs.get(&(host.to_string(), !http2)) {
+                let mut config = ServerConfig::clone(other);
+                config.alpn_protocols = alpn(http2);
+                let config = Arc::new(config);
+                configs.insert(key, config.clone());
+                return Ok(config);
+            }
         }
 
+        let mut config = sign()?;
+        config.alpn_protocols = alpn(http2);
+        let mut configs = self.leaf_configs.lock().unwrap();
+        Ok(configs.entry(key).or_insert_with(|| Arc::new(config)).clone())
+    }
+
+    /// A TLS config with a new leaf for `host`, signed by the CA. Key generation and signing
+    /// take a few milliseconds.
+    fn sign_leaf(&self, host: &str) -> Result<ServerConfig, BoxError> {
         let now = OffsetDateTime::now_utc();
         let mut params = CertificateParams::new(vec![host.to_string()])?;
         params.distinguished_name.push(DnType::CommonName, host);
@@ -115,22 +152,22 @@ impl CaStore {
         let leaf_key = KeyPair::generate()?;
         let leaf_cert = params.signed_by(&leaf_key, &self.issuer, &self.key)?;
 
-        let mut config =
-            ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
-                .with_safe_default_protocol_versions()?
-                .with_no_client_auth()
-                .with_single_cert(
-                    vec![leaf_cert.der().clone(), self.cert_der.clone()],
-                    PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(leaf_key.serialize_der())),
-                )?;
-        config.alpn_protocols = vec![b"http/1.1".to_vec()];
-
-        let config = Arc::new(config);
-        self.leaf_configs.lock().unwrap().insert(host.to_string(), config.clone());
-        Ok(config)
+        Ok(ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
+            .with_safe_default_protocol_versions()?
+            .with_no_client_auth()
+            .with_single_cert(
+                vec![leaf_cert.der().clone(), self.cert_der.clone()],
+                PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(leaf_key.serialize_der())),
+            )?)
     }
 }
 
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+fn alpn(http2: bool) -> Vec<Vec<u8>> {
+    if http2 { vec![b"h2".to_vec(), b"http/1.1".to_vec()] } else { vec![b"http/1.1".to_vec()] }
+}
+
 /// Issuer fields for the CA. Leaf certificates are signed against these, so they must stay
 /// identical to what was written to `ca.pem`.
 fn ca_params() -> CertificateParams {
@@ -173,3 +210,45 @@ fn write_atomic(path: &Path, contents: &[u8], private: bool) -> io::Result<()> {
     file.sync_all()?;
     fs::rename(&tmp, path)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::mpsc;
+
+    /// A host whose leaf is still being signed holds up no lookup of another host's.
+    #[test]
+    fn signing_one_host_does_not_block_another() {
+        let dir = tempfile::tempdir().unwrap();
+        let ca = Arc::new(CaStore::load_or_create(dir.path()).unwrap());
+        let cached = ca.server_config("cached.test", true).unwrap();
+
+        let (signing_tx, signing_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel::<()>();
+        let slow = std::thread::spawn({
+            let ca = ca.clone();
+            move || {
+                ca.server_config_with("slow.test", true, || {
+                    signing_tx.send(()).unwrap();
+                    release_rx.recv().unwrap();
+                    ca.sign_leaf("slow.test")
+                })
+            }
+        });
+        signing_rx.recv().unwrap();
+
+        let (found_tx, found_rx) = mpsc::channel();
+        std::thread::spawn({
+            let ca = ca.clone();
+            move || found_tx.send(ca.server_config("cached.test", true).unwrap())
+        });
+        let found = found_rx.recv_timeout(std::time::Duration::from_secs(5));
+        release_tx.send(()).unwrap();
+        let found = found.expect("the cached lookup waited for another host's signing");
+        assert!(Arc::ptr_eq(&found, &cached));
+
+        let slow = slow.join().unwrap().unwrap();
+        assert_eq!(slow.alpn_protocols, alpn(true));
+        assert!(Arc::ptr_eq(&slow, &ca.server_config("slow.test", true).unwrap()), "kept");
+    }
+}

crates/yaak-proxy/src/connection.rs

@@ -2,13 +2,30 @@ use std::sync::Arc;
 
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
-use hyper_util::rt::TokioIo;
+use hyper_util::rt::{TokioExecutor, TokioIo};
+use hyper_util::server::conn::auto;
 use tokio::io::{AsyncRead, AsyncWrite};
 
 use crate::request::handle_request;
 use crate::{ClientConn, Engine};
 
-pub(crate) async fn serve_http1<S, F, Fut>(io: S, handler: F) -> Result<(), hyper::Error>
+/// Streams one HTTP/2 client connection may have open at once.
+const MAX_CONCURRENT_STREAMS: u32 = 100;
+/// What an HTTP/2 client may send before the engine has forwarded it, per stream and in all.
+/// The same as hyper's client uses upstream, so uploads are not throttled in between.
+const STREAM_WINDOW: u32 = 2 * 1024 * 1024;
+const CONNECTION_WINDOW: u32 = 5 * 1024 * 1024;
+/// Request headers an HTTP/2 client may send, cookies included. hyper's 16 KB default is less
+/// than a browser sends some sites.
+const MAX_HEADER_LIST_SIZE: u32 = 256 * 1024;
+
+/// Serves a decrypted client connection in the protocol it chose in ALPN: HTTP/2, with each
+/// stream a request of its own, or HTTP/1.1 as on the plain listener.
+pub(crate) async fn serve_tls<S, F, Fut>(
+    io: S,
+    http2: bool,
+    handler: F,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
 where
     S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     F: Fn(hyper::Request<hyper::body::Incoming>) -> Fut + Send + Sync + 'static,
@@ -16,18 +33,31 @@ where
         + Send
         + 'static,
 {
-    http1::Builder::new()
-        .preserve_header_case(true)
-        .title_case_headers(true)
-        .serve_connection(TokioIo::new(io), service_fn(handler))
-        .with_upgrades()
-        .await
+    let mut builder = auto::Builder::new(TokioExecutor::new());
+    builder.http1().preserve_header_case(true).title_case_headers(true);
+    builder
+        .http2()
+        .max_concurrent_streams(MAX_CONCURRENT_STREAMS)
+        .initial_stream_window_size(STREAM_WINDOW)
+        .initial_connection_window_size(CONNECTION_WINDOW)
+        .max_header_list_size(MAX_HEADER_LIST_SIZE);
+    let builder = if http2 { builder.http2_only() } else { builder.http1_only() };
+    builder.serve_connection_with_upgrades(TokioIo::new(io), service_fn(handler)).await
 }
 
+/// Serves a connection to the listener, in HTTP/1.1 only: HTTP/2 is offered inside TLS alone.
 pub(crate) async fn handle_connection(
     stream: tokio::net::TcpStream,
     client: ClientConn,
     engine: Arc<Engine>,
 ) -> Result<(), hyper::Error> {
-    serve_http1(stream, move |req| handle_request(req, client, engine.clone())).await
+    http1::Builder::new()
+        .preserve_header_case(true)
+        .title_case_headers(true)
+        .serve_connection(
+            TokioIo::new(stream),
+            service_fn(move |req| handle_request(req, client, engine.clone())),
+        )
+        .with_upgrades()
+        .await
 }

crates/yaak-proxy/src/lib.rs

@@ -8,7 +8,7 @@ use std::collections::hash_map::RandomState;
 use std::collections::{HashMap, HashSet};
 use std::hash::{BuildHasher, Hasher};
 use std::net::SocketAddr;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::mpsc as std_mpsc;
 use std::sync::{Arc, Mutex};
 use std::time::SystemTime;
@@ -46,6 +46,9 @@ pub struct ProxyConfig {
     /// number and the client's address, before anything is read from it, so the client is
     /// still there to be looked up. The connection waits for it.
     pub on_accept: Option<AcceptHook>,
+    /// Offer HTTP/2 to clients in intercepted TLS, beside HTTP/1.1. Off, they all speak
+    /// HTTP/1.1, for clients that misbehave on HTTP/2. See [`ProxyHandle::set_http2`].
+    pub http2: bool,
 }
 
 /// See [`ProxyConfig::on_accept`].
@@ -59,14 +62,16 @@ impl ProxyConfig {
             capture_body_limit: DEFAULT_CAPTURE_BODY_LIMIT,
             upstream_root_certs: Vec::new(),
             on_accept: None,
+            http2: true,
         }
     }
 }
 
 /// Lifecycle of each proxied exchange (a flow), keyed by `id`. A flow starts with
-/// `RequestStart` and ends with exactly one `ResponseBodyComplete` or `Error`. Traffic the
-/// engine tunnels without decrypting (pinned clients, non-TLS CONNECT payloads, the frames
-/// after a `101 Switching Protocols`) produces no events.
+/// `RequestStart` and ends with exactly one `ResponseBodyComplete`, `Error` or `Cancelled`.
+/// Each HTTP/2 stream is a flow of its own. Traffic the engine tunnels without decrypting
+/// (pinned clients, non-TLS CONNECT payloads, the frames after a `101 Switching Protocols`)
+/// produces no events.
 #[derive(Debug, Clone)]
 pub enum ProxyEvent {
     /// Request headers arrived from the client.
@@ -125,7 +130,7 @@ pub enum ProxyEvent {
         /// The captured part of the frame; empty once `capture_body_limit` is reached.
         data: Bytes,
     },
-    /// The response body ended, normally or with `error` (upstream failed or client left).
+    /// The response body ended: normally, with `error` when upstream failed, or `cancelled`.
     ResponseBodyComplete {
         id: u64,
         size: u64,
@@ -134,6 +139,9 @@ pub enum ProxyEvent {
         /// Milliseconds from `started_at` to the end of the body.
         elapsed_ms: u64,
         error: Option<String>,
+        /// The client stopped reading before the body ended: it reset the stream or closed the
+        /// connection. Not an error of the engine's or the server's.
+        cancelled: bool,
     },
     /// No response: connecting to or requesting from upstream failed, and the client got a 502,
     /// or the engine refused the request (no host to send it to, or it looped back to the engine)
@@ -143,6 +151,9 @@ pub enum ProxyEvent {
         error: String,
         elapsed_ms: u64,
     },
+    /// The client gave up before the response headers arrived, by resetting the stream or
+    /// closing the connection. The request upstream is cancelled with it.
+    Cancelled { id: u64, elapsed_ms: u64 },
 }
 
 /// A running engine. Dropping it stops the listener and joins its thread.
@@ -168,6 +179,12 @@ impl ProxyHandle {
         self.engine.handshake_failures.lock().unwrap().clear();
     }
 
+    /// Offers HTTP/2 to clients, or stops, from the next connection on. Connections already
+    /// open keep the protocol they chose.
+    pub fn set_http2(&self, enabled: bool) {
+        self.engine.http2.store(enabled, Ordering::Relaxed);
+    }
+
     /// Handshake failures counted toward passing `addr` (`host:port`) through, and whether it
     /// already is. For tests that must wait for the engine to record a failure.
     #[doc(hidden)]
@@ -206,6 +223,8 @@ pub(crate) struct Engine {
     upstream_tls_h1: Arc<ClientConfig>,
     passthrough: Mutex<HashSet<String>>,
     handshake_failures: Mutex<HashMap<String, u32>>,
+    /// Offer HTTP/2 to clients in intercepted TLS.
+    http2: AtomicBool,
     /// `Via` value marking requests whose URL the engine rebuilt from `Host`, unique to this run,
     /// so one that comes back to the engine is refused instead of forwarded again.
     via: HeaderValue,
@@ -220,6 +239,10 @@ impl Engine {
         self.next_id.fetch_add(1, Ordering::Relaxed)
     }
 
+    fn offers_http2(&self) -> bool {
+        self.http2.load(Ordering::Relaxed)
+    }
+
     fn is_passthrough(&self, addr: &str) -> bool {
         self.passthrough.lock().unwrap().contains(addr)
     }
@@ -284,6 +307,7 @@ pub fn start_proxy(config: ProxyConfig) -> Result<ProxyHandle, String> {
         upstream_tls_h1: upstream_tls_config(&roots, &[b"http/1.1"])?,
         passthrough: Mutex::new(HashSet::new()),
         handshake_failures: Mutex::new(HashMap::new()),
+        http2: AtomicBool::new(config.http2),
         via: via_marker(),
     });
 

crates/yaak-proxy/src/request.rs

@@ -2,12 +2,18 @@ use std::convert::Infallible;
 use std::error::Error;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, Instant, SystemTime};
 
 use bytes::Bytes;
 use http_body_util::{BodyExt, Full};
+use hyper::body::Body;
 use hyper::body::Incoming;
-use hyper::header::{CONTENT_TYPE, HOST, UPGRADE, VIA};
+use hyper::client::conn::TrySendError;
+use hyper::header::{
+    CONTENT_LENGTH, CONTENT_TYPE, COOKIE, HOST, HeaderValue, TRANSFER_ENCODING, UPGRADE, VIA,
+};
+use hyper::http::request::Parts;
 use hyper::http::uri::{Authority, PathAndQuery, Scheme};
 use hyper::upgrade::OnUpgrade;
 use hyper::{Method, Request, Response, StatusCode, Uri};
@@ -20,7 +26,7 @@ use tokio::net::TcpStream;
 use tokio_rustls::{LazyConfigAcceptor, TlsConnector};
 
 use crate::body::{CaptureBody, Direction};
-use crate::connection::serve_http1;
+use crate::connection::serve_tls;
 use crate::tunnel::{Rewind, splice, tunnel};
 use crate::{ClientConn, Engine, ProxyEvent};
 
@@ -64,6 +70,10 @@ struct Flow {
     id: u64,
     start: Instant,
     engine: Arc<Engine>,
+    /// It has answered the client, so dropping it is no cancel.
+    answered: bool,
+    /// Reading the client's upload failed: it reset the stream or closed the connection.
+    upload_failed: Arc<AtomicBool>,
 }
 
 impl Flow {
@@ -73,7 +83,13 @@ impl Flow {
         req: &Request<Incoming>,
         url: String,
     ) -> Self {
-        let flow = Self { id: engine.next_flow_id(), start: Instant::now(), engine };
+        let flow = Self {
+            id: engine.next_flow_id(),
+            start: Instant::now(),
+            engine,
+            answered: false,
+            upload_failed: Arc::new(AtomicBool::new(false)),
+        };
         flow.engine.emit(ProxyEvent::RequestStart {
             id: flow.id,
             started_at: SystemTime::now(),
@@ -83,6 +99,18 @@ impl Flow {
             connection: client.connection,
             client_addr: client.addr,
         });
+        // HTTP/2 carries the host in `:authority`, which hyper keeps in the URI. Shown as `Host`,
+        // as HTTP/1.1 requests have it.
+        if req.version() == hyper::Version::HTTP_2
+            && !req.headers().contains_key(HOST)
+            && let Some(authority) = req.uri().authority()
+        {
+            flow.engine.emit(ProxyEvent::RequestHeader {
+                id: flow.id,
+                name: HOST.to_string(),
+                value: authority.to_string(),
+            });
+        }
         for (name, value) in req.headers() {
             flow.engine.emit(ProxyEvent::RequestHeader {
                 id: flow.id,
@@ -94,14 +122,17 @@ impl Flow {
     }
 
     fn capture_request(&self, body: Incoming) -> BoxBody {
-        CaptureBody::new(body, Direction::Request, self.id, self.start, self.engine.clone()).boxed()
+        CaptureBody::new(body, Direction::Request, self.id, self.start, self.engine.clone())
+            .flag_failure(self.upload_failed.clone())
+            .boxed()
     }
 
     fn respond(
-        self,
+        mut self,
         resp: Response<Incoming>,
         remote_addr: Option<SocketAddr>,
     ) -> Response<BoxBody> {
+        self.answered = true;
         self.engine.emit(ProxyEvent::ResponseStart {
             id: self.id,
             status: resp.status().as_u16(),
@@ -117,12 +148,14 @@ impl Flow {
             });
         }
         let (parts, body) = resp.into_parts();
-        let body = CaptureBody::new(body, Direction::Response, self.id, self.start, self.engine);
+        let engine = self.engine.clone();
+        let body = CaptureBody::new(body, Direction::Response, self.id, self.start, engine);
         Response::from_parts(parts, body.boxed())
     }
 
     /// Answers the client with `status` and `error` without contacting any server.
-    fn reject(self, status: StatusCode, error: &str) -> Response<BoxBody> {
+    fn reject(mut self, status: StatusCode, error: &str) -> Response<BoxBody> {
+        self.answered = true;
         debug!("Flow {} rejected: {error}", self.id);
         self.engine.emit(ProxyEvent::Error {
             id: self.id,
@@ -136,7 +169,14 @@ impl Flow {
             .unwrap()
     }
 
-    fn fail(self, error: String) -> Response<BoxBody> {
+    fn fail(mut self, error: String) -> Response<BoxBody> {
+        // The request failed because the client stopped sending its body: it cancelled. The
+        // answer goes nowhere, and dropping the flow unanswered says so.
+        if self.upload_failed.load(Ordering::Relaxed) {
+            debug!("Flow {} upload ended by the client: {error}", self.id);
+            return bad_gateway(error);
+        }
+        self.answered = true;
         debug!("Flow {} failed: {error}", self.id);
         let body = format!("Yaak Proxy could not reach the server: {error}");
         self.engine.emit(ProxyEvent::Error {
@@ -148,6 +188,20 @@ impl Flow {
     }
 }
 
+/// A flow dropped before it answered was abandoned by the client, which reset its stream or
+/// closed the connection: hyper drops the request's future, and the request upstream with it.
+impl Drop for Flow {
+    fn drop(&mut self) {
+        if !self.answered {
+            debug!("Flow {} cancelled by the client", self.id);
+            self.engine.emit(ProxyEvent::Cancelled {
+                id: self.id,
+                elapsed_ms: self.start.elapsed().as_millis() as u64,
+            });
+        }
+    }
+}
+
 fn bad_gateway(body: String) -> Response<BoxBody> {
     Response::builder().status(StatusCode::BAD_GATEWAY).body(text_body(body)).unwrap()
 }
@@ -296,7 +350,7 @@ where
         .server_name()
         .map(str::to_string)
         .unwrap_or_else(|| target.server_name.clone());
-    let config = engine.ca.server_config(&server_name)?;
+    let config = engine.ca.server_config(&server_name, engine.offers_http2())?;
     let mut tls = match start.into_stream(config).await {
         Ok(tls) => tls,
         Err(e) => {
@@ -323,10 +377,11 @@ where
     buf.truncate(n);
     engine.record_handshake_success(&target.addr);
 
+    let http2 = tls.get_ref().1.alpn_protocol() == Some(b"h2");
     let tls = Rewind::new(Bytes::from(buf), tls);
-    let target = Arc::new(Target { server_name, ..target });
-    serve_http1(tls, move |req| forward_https(req, target.clone(), conn, engine.clone())).await?;
-    Ok(())
+    let upstreams = Arc::new(Upstreams::new(Target { server_name, ..target }));
+    serve_tls(tls, http2, move |req| forward_https(req, upstreams.clone(), conn, engine.clone()))
+        .await
 }
 
 fn record_handshake_failure(
@@ -371,11 +426,115 @@ enum Upstream {
     H2(hyper::client::conn::http2::SendRequest<BoxBody>),
 }
 
+/// A connection to the server, and the address it reached.
+struct UpstreamConn {
+    sender: Upstream,
+    remote_addr: Option<SocketAddr>,
+}
+
+impl UpstreamConn {
+    fn is_h2(&self) -> bool {
+        matches!(self.sender, Upstream::H2(_))
+    }
+
+    fn is_closed(&self) -> bool {
+        match &self.sender {
+            Upstream::H1(s) => s.is_closed(),
+            Upstream::H2(s) => s.is_closed(),
+        }
+    }
+
+    /// Can take a request now: an HTTP/2 connection any time it is open, an HTTP/1.1 one once
+    /// the response before has been read.
+    fn is_ready(&self) -> bool {
+        match &self.sender {
+            Upstream::H1(s) => s.is_ready(),
+            Upstream::H2(s) => s.is_ready(),
+        }
+    }
+
+    async fn send(
+        &mut self,
+        req: Request<BoxBody>,
+    ) -> Result<Response<Incoming>, TrySendError<Request<BoxBody>>> {
+        match &mut self.sender {
+            Upstream::H1(s) => s.try_send_request(req).await,
+            Upstream::H2(s) => s.try_send_request(req).await,
+        }
+    }
+}
+
+/// Idle HTTP/1.1 connections kept per client connection, for its requests to come.
+const MAX_IDLE_H1: usize = 6;
+
+/// The connections to the server of one intercepted client connection, reused by its requests,
+/// so an HTTP/2 client's streams do not each pay for a handshake. An HTTP/2 connection carries
+/// them all at once; an HTTP/1.1 one a request at a time.
+struct Upstreams {
+    target: Target,
+    conns: tokio::sync::Mutex<Vec<UpstreamConn>>,
+    /// The server answered in HTTP/1.1, so each request in flight needs a connection of its
+    /// own, and new ones need not wait to see whether the one before can be shared.
+    h1: AtomicBool,
+}
+
+impl Upstreams {
+    fn new(target: Target) -> Self {
+        Self { target, conns: tokio::sync::Mutex::new(Vec::new()), h1: AtomicBool::new(false) }
+    }
+
+    /// A connection for one request, and whether it carried one before. Requests that arrive
+    /// before the first connection is up wait for it, to share it if it is HTTP/2.
+    async fn take(&self, engine: &Engine) -> Result<(UpstreamConn, bool), BoxError> {
+        let mut conns = self.conns.lock().await;
+        conns.retain(|c| !c.is_closed());
+        if let Some(i) = conns.iter().position(UpstreamConn::is_ready) {
+            let conn = match &conns[i].sender {
+                Upstream::H2(s) => UpstreamConn {
+                    sender: Upstream::H2(s.clone()),
+                    remote_addr: conns[i].remote_addr,
+                },
+                Upstream::H1(_) => conns.swap_remove(i),
+            };
+            return Ok((conn, true));
+        }
+        if self.h1.load(Ordering::Relaxed) {
+            drop(conns);
+            return Ok((connect_upstream(engine, &self.target, false).await?, false));
+        }
+        let conn = connect_upstream(engine, &self.target, false).await?;
+        match &conn.sender {
+            Upstream::H2(s) => conns.push(UpstreamConn {
+                sender: Upstream::H2(s.clone()),
+                remote_addr: conn.remote_addr,
+            }),
+            Upstream::H1(_) => self.h1.store(true, Ordering::Relaxed),
+        }
+        Ok((conn, false))
+    }
+
+    /// Keeps a connection for later requests. An HTTP/1.1 one takes the next once this response
+    /// has been read. An HTTP/2 one is kept already, unless it was opened to send a request again.
+    async fn put_back(&self, conn: UpstreamConn) {
+        if conn.is_closed() {
+            return;
+        }
+        let mut conns = self.conns.lock().await;
+        if conn.is_h2() {
+            if !conns.iter().any(|c| c.is_h2() && !c.is_closed()) {
+                conns.push(conn);
+            }
+        } else if conns.len() < MAX_IDLE_H1 {
+            conns.push(conn);
+        }
+    }
+}
+
 async fn connect_upstream(
     engine: &Engine,
     target: &Target,
     h1_only: bool,
-) -> Result<(Upstream, Option<SocketAddr>), BoxError> {
+) -> Result<UpstreamConn, BoxError> {
     let tcp = TcpStream::connect(&target.addr).await?;
     let remote_addr = tcp.peer_addr().ok();
     let config = if h1_only { &engine.upstream_tls_h1 } else { &engine.upstream_tls };
@@ -406,61 +565,118 @@ async fn connect_upstream(
         });
         Upstream::H1(sender)
     };
-    Ok((upstream, remote_addr))
+    Ok(UpstreamConn { sender: upstream, remote_addr })
+}
+
+/// Addresses a request to the server in the protocol of `conn`. It may be addressed for the
+/// other one already, when it is sent again on a new connection.
+fn address(
+    parts: &mut Parts,
+    conn: &UpstreamConn,
+    target: &Target,
+    path: &str,
+    has_body: bool,
+) -> Result<(), String> {
+    // The host the client asked for: `Host` over HTTP/1.1, `:authority` over HTTP/2, which
+    // hyper keeps in the URI.
+    let authority =
+        parts.uri.authority().map_or_else(|| target.authority.clone(), |a| a.to_string());
+    if conn.is_h2() {
+        // `Host` moves into `:authority`. Google's servers reset any stream that has both, even
+        // when they match.
+        let host = match parts.headers.remove(HOST) {
+            Some(host) => host.to_str().map_err(|e| e.to_string())?.to_string(),
+            None => authority,
+        };
+        parts.uri = format!("https://{host}{path}").parse::<Uri>().map_err(|e| e.to_string())?;
+    } else {
+        parts.uri = path.parse::<Uri>().map_err(|e| e.to_string())?;
+        if !parts.headers.contains_key(HOST)
+            && let Ok(host) = authority.parse()
+        {
+            parts.headers.insert(HOST, host);
+        }
+        if parts.version == hyper::Version::HTTP_2 {
+            parts.version = hyper::Version::HTTP_11;
+            // HTTP/2 frames a body without a length. HTTP/1.1 needs one or chunks, and hyper
+            // sends a GET with neither as empty. Chunks also carry trailers, those named in
+            // `Trailer`: hyper's HTTP/1.1 encoder drops any others.
+            if has_body && !parts.headers.contains_key(CONTENT_LENGTH) {
+                parts.headers.insert(TRANSFER_ENCODING, HeaderValue::from_static("chunked"));
+            }
+            // HTTP/2 may split `Cookie` into a field per cookie; HTTP/1.1 has one (RFC 9113
+            // 8.2.3).
+            if parts.headers.get_all(COOKIE).iter().nth(1).is_some() {
+                let joined = parts
+                    .headers
+                    .get_all(COOKIE)
+                    .iter()
+                    .map(HeaderValue::as_bytes)
+                    .collect::<Vec<_>>()
+                    .join(&b"; "[..]);
+                let joined = HeaderValue::from_bytes(&joined).map_err(|e| e.to_string())?;
+                parts.headers.insert(COOKIE, joined);
+            }
+        }
+    }
+    Ok(())
 }
 
 async fn forward_https(
     mut req: Request<Incoming>,
-    target: Arc<Target>,
+    upstreams: Arc<Upstreams>,
     client: ClientConn,
     engine: Arc<Engine>,
 ) -> Result<Response<BoxBody>, Infallible> {
+    let target = &upstreams.target;
     let path = req.uri().path_and_query().map(|pq| pq.to_string()).unwrap_or_else(|| "/".into());
     let url = format!("https://{}{path}", target.authority);
-    let flow = Flow::begin(engine.clone(), client, &req, url.clone());
+    let flow = Flow::begin(engine.clone(), client, &req, url);
 
+    // Only HTTP/1.1 has `Upgrade`. Its connection is the tunnel afterwards, so it is not shared.
     let client_upgrade: Option<OnUpgrade> =
         req.headers().contains_key(UPGRADE).then(|| hyper::upgrade::on(&mut req));
-
-    let (mut upstream, remote_addr) =
-        match connect_upstream(&engine, &target, client_upgrade.is_some()).await {
-            Ok(u) => u,
-            Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))),
-        };
+    let taken = match client_upgrade {
+        Some(_) => connect_upstream(&engine, target, true).await.map(|c| (c, false)),
+        None => upstreams.take(&engine).await,
+    };
+    let (mut conn, mut reused) = match taken {
+        Ok(taken) => taken,
+        Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))),
+    };
 
     let (mut parts, body) = req.into_parts();
-    let uri = match &upstream {
-        // `Host` moves into `:authority`. Google's servers reset any stream that has both, even
-        // when they match.
-        Upstream::H2(_) => match parts.headers.remove(HOST) {
-            Some(host) => match host.to_str() {
-                Ok(host) => format!("https://{host}{path}").parse::<Uri>(),
-                Err(e) => return Ok(flow.fail(e.to_string())),
-            },
-            None => url.parse::<Uri>(),
-        },
-        Upstream::H1(_) => path.parse::<Uri>(),
-    };
-    parts.uri = match uri {
-        Ok(uri) => uri,
-        Err(e) => return Ok(flow.fail(e.to_string())),
-    };
-    if matches!(upstream, Upstream::H1(_))
-        && !parts.headers.contains_key(HOST)
-        && let Ok(host) = target.authority.parse()
-    {
-        parts.headers.insert(HOST, host);
+    if let Err(e) = address(&mut parts, &conn, target, &path, !body.is_end_stream()) {
+        return Ok(flow.fail(e));
     }
-    let outgoing = Request::from_parts(parts, flow.capture_request(body));
+    let mut outgoing = Request::from_parts(parts, flow.capture_request(body));
 
-    let sent = match &mut upstream {
-        Upstream::H1(s) => s.send_request(outgoing).await,
-        Upstream::H2(s) => s.send_request(outgoing).await,
-    };
-    let mut resp = match sent {
-        Ok(resp) => resp,
-        Err(e) => return Ok(flow.fail(error_chain(&e))),
+    let mut resp = loop {
+        let mut e = match conn.send(outgoing).await {
+            Ok(resp) => break resp,
+            Err(e) => e,
+        };
+        // A reused connection the server closed as the request went out: the request never
+        // left, so it goes again, once, on a new connection.
+        let Some(unsent) = e.take_message().filter(|_| reused) else {
+            return Ok(flow.fail(error_chain(e.error())));
+        };
+        debug!("Upstream connection to {} was closed, retrying on a new one", target.addr);
+        conn = match connect_upstream(&engine, target, false).await {
+            Ok(conn) => conn,
+            Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))),
+        };
+        reused = false;
+        let (mut parts, body) = unsent.into_parts();
+        if let Err(e) = address(&mut parts, &conn, target, &path, !body.is_end_stream()) {
+            return Ok(flow.fail(e));
+        }
+        outgoing = Request::from_parts(parts, body);
     };
+    let remote_addr = conn.remote_addr;
+    if client_upgrade.is_none() {
+        upstreams.put_back(conn).await;
+    }
 
     if resp.status() == StatusCode::SWITCHING_PROTOCOLS
         && let Some(client_upgrade) = client_upgrade

crates/yaak-proxy/tests/proxy.rs

@@ -1,6 +1,8 @@
+use std::collections::{HashMap, HashSet};
 use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::mpsc::Receiver;
 use std::time::{Duration, SystemTime};
 
@@ -9,9 +11,10 @@ use futures_util::{SinkExt, StreamExt};
 use http_body_util::combinators::BoxBody;
 use http_body_util::{BodyExt, Full, StreamBody};
 use hyper::body::{Frame, Incoming};
-use hyper::header::{CONNECTION, CONTENT_LENGTH, CONTENT_TYPE, HOST, UPGRADE};
+use hyper::client::conn::http2;
+use hyper::header::{CONNECTION, CONTENT_LENGTH, CONTENT_TYPE, HOST, HeaderValue, UPGRADE};
 use hyper::service::service_fn;
-use hyper::{Request, Response, StatusCode};
+use hyper::{Request, Response, StatusCode, Version};
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use hyper_util::server::conn::auto;
 use rcgen::{BasicConstraints, CertificateParams, IsCa, KeyPair};
@@ -86,7 +89,11 @@ fn full(data: impl Into<Bytes>) -> TestBody {
 async fn route(req: Request<Incoming>) -> Result<Response<TestBody>, Infallible> {
     Ok(match req.uri().path() {
         "/hello" => Response::new(full("hello")),
-        "/echo" => Response::new(full(req.into_body().collect().await.unwrap().to_bytes())),
+        // An upload cut short, as a cancelled one is, gets an empty answer.
+        "/echo" => match req.into_body().collect().await {
+            Ok(body) => Response::new(full(body.to_bytes())),
+            Err(_) => Response::builder().status(StatusCode::BAD_REQUEST).body(full("")).unwrap(),
+        },
         "/authority" => Response::new(full(format!(
             "authority={} host={}",
             req.uri().authority().map_or("none".into(), |a| a.to_string()),
@@ -97,10 +104,40 @@ async fn route(req: Request<Incoming>) -> Result<Response<TestBody>, Infallible>
         // The stream ends with half the promised bytes, so the upstream resets it mid-body.
         "/cut" => late_end(10),
         "/ws" => websocket_echo(req),
+        "/trailers" => echo_with_trailers(req).await,
+        "/cookies" => {
+            let cookies = req.headers().get_all("cookie").iter();
+            Response::new(full(
+                cookies.map(|c| c.to_str().unwrap()).collect::<Vec<_>>().join(" | "),
+            ))
+        }
+        // Headers for this connection alone, which HTTP/2 forbids.
+        "/hop" => Response::builder()
+            .header(CONNECTION, "x-hop")
+            .header("x-hop", "1")
+            .header("keep-alive", "timeout=5")
+            .body(full("hop"))
+            .unwrap(),
         _ => Response::builder().status(StatusCode::NOT_FOUND).body(full("not found")).unwrap(),
     })
 }
 
+/// The request body back, with the request's trailers and one of the server's own.
+async fn echo_with_trailers(req: Request<Incoming>) -> Response<TestBody> {
+    let collected = req.into_body().collect().await.unwrap();
+    let mut trailers = collected.trailers().cloned().unwrap_or_default();
+    trailers.insert("x-upstream-trailer", HeaderValue::from_static("yes"));
+    let frames = [
+        Ok(Frame::data(collected.to_bytes())),
+        Ok(Frame::trailers(trailers)),
+    ];
+    Response::builder()
+        // HTTP/1.1 sends only the trailers it names.
+        .header("trailer", "x-client-trailer, x-upstream-trailer")
+        .body(BodyExt::boxed(StreamBody::new(futures_util::stream::iter(frames))))
+        .unwrap()
+}
+
 /// "hello" under a Content-Length of `content_length`, its end of stream sent a while after the
 /// last byte, as an HTTP/2 server does when its final DATA frame has no END_STREAM.
 fn late_end(content_length: usize) -> Response<TestBody> {
@@ -139,15 +176,31 @@ fn websocket_echo(mut req: Request<Incoming>) -> Response<TestBody> {
 }
 
 async fn spawn_upstream(tls: Option<Arc<ServerConfig>>) -> SocketAddr {
+    spawn_upstream_with(tls, Arc::new(AtomicUsize::new(0)), route).await
+}
+
+/// An upstream server answering with `handler`, counting the connections it accepts in
+/// `accepts`.
+async fn spawn_upstream_with<F, Fut>(
+    tls: Option<Arc<ServerConfig>>,
+    accepts: Arc<AtomicUsize>,
+    handler: F,
+) -> SocketAddr
+where
+    F: Fn(Request<Incoming>) -> Fut + Clone + Send + Sync + 'static,
+    Fut: Future<Output = Result<Response<TestBody>, Infallible>> + Send + 'static,
+{
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
     tokio::spawn(async move {
         loop {
             let (tcp, _) = listener.accept().await.unwrap();
+            accepts.fetch_add(1, Ordering::SeqCst);
             let tls = tls.clone();
+            let handler = handler.clone();
             tokio::spawn(async move {
                 let builder = auto::Builder::new(TokioExecutor::new());
-                let svc = service_fn(route);
+                let svc = service_fn(handler);
                 let _ = match tls {
                     Some(config) => {
                         let Ok(stream) = TlsAcceptor::from(config).accept(tcp).await else {
@@ -193,8 +246,10 @@ impl Harness {
         format!("http://127.0.0.1:{}", self.proxy.port)
     }
 
+    /// An HTTP/1.1 client. The HTTP/2 tests below make their own.
     fn client_trusting(&self, root_pem: &str) -> reqwest::Client {
         reqwest::Client::builder()
+            .http1_only()
             .proxy(reqwest::Proxy::all(self.proxy_url()).unwrap())
             .add_root_certificate(reqwest::Certificate::from_pem(root_pem.as_bytes()).unwrap())
             .build()
@@ -261,8 +316,7 @@ impl Harness {
         loop {
             let event =
                 self.events.recv_timeout(Duration::from_secs(10)).expect("flow did not end");
-            let done =
-                matches!(event, ProxyEvent::ResponseBodyComplete { .. } | ProxyEvent::Error { .. });
+            let done = is_terminal(&event);
             events.push(event);
             if done {
                 break;
@@ -273,6 +327,87 @@ impl Harness {
         }
         events
     }
+
+    /// Events of the next `n` flows, which may interleave, by flow id.
+    fn flows_events(&self, n: usize) -> HashMap<u64, Vec<ProxyEvent>> {
+        let mut flows: HashMap<u64, Vec<ProxyEvent>> = HashMap::new();
+        let mut ended = 0;
+        while ended < n {
+            let event =
+                self.events.recv_timeout(Duration::from_secs(10)).expect("flows did not end");
+            ended += usize::from(is_terminal(&event));
+            flows.entry(flow_id(&event)).or_default().push(event);
+        }
+        while let Ok(event) = self.events.recv_timeout(Duration::from_millis(100)) {
+            flows.entry(flow_id(&event)).or_default().push(event);
+        }
+        flows
+    }
+
+    /// TLS to `localhost:port` through a CONNECT tunnel, trusting the proxy's CA and offering
+    /// `alpn`.
+    async fn tls_tunnel(
+        &self,
+        port: u16,
+        alpn: &[&[u8]],
+    ) -> tokio_rustls::client::TlsStream<TcpStream> {
+        let mut config = ClientConfig::clone(&self.tls_client_config());
+        config.alpn_protocols = alpn.iter().map(|p| p.to_vec()).collect();
+        let tcp = self.connect_tunnel(port).await;
+        let server_name = ServerName::try_from("localhost").unwrap();
+        TlsConnector::from(Arc::new(config)).connect(server_name, tcp).await.unwrap()
+    }
+
+    /// An HTTP/2 connection to `localhost:port` through the proxy, offering both versions as a
+    /// browser does.
+    async fn h2_client(&self, port: u16) -> http2::SendRequest<TestBody> {
+        let tls = self.tls_tunnel(port, &[b"h2", b"http/1.1"]).await;
+        assert_eq!(tls.get_ref().1.alpn_protocol(), Some(&b"h2"[..]));
+        let (sender, conn) =
+            http2::handshake(TokioExecutor::new(), TokioIo::new(tls)).await.unwrap();
+        tokio::spawn(conn);
+        sender
+    }
+}
+
+fn is_terminal(event: &ProxyEvent) -> bool {
+    matches!(
+        event,
+        ProxyEvent::ResponseBodyComplete { .. }
+            | ProxyEvent::Error { .. }
+            | ProxyEvent::Cancelled { .. }
+    )
+}
+
+fn flow_id(event: &ProxyEvent) -> u64 {
+    match event {
+        ProxyEvent::RequestStart { id, .. }
+        | ProxyEvent::RequestHeader { id, .. }
+        | ProxyEvent::RequestBodyChunk { id, .. }
+        | ProxyEvent::RequestBodyComplete { id, .. }
+        | ProxyEvent::ResponseStart { id, .. }
+        | ProxyEvent::ResponseHeader { id, .. }
+        | ProxyEvent::ResponseBodyChunk { id, .. }
+        | ProxyEvent::ResponseBodyComplete { id, .. }
+        | ProxyEvent::Error { id, .. }
+        | ProxyEvent::Cancelled { id, .. } => *id,
+    }
+}
+
+/// The headers a flow captured on one side, in order.
+fn headers(events: &[ProxyEvent], request: bool) -> Vec<(String, String)> {
+    events
+        .iter()
+        .filter_map(|e| match e {
+            ProxyEvent::RequestHeader { name, value, .. } if request => {
+                Some((name.clone(), value.clone()))
+            }
+            ProxyEvent::ResponseHeader { name, value, .. } if !request => {
+                Some((name.clone(), value.clone()))
+            }
+            _ => None,
+        })
+        .collect()
 }
 
 /// The captured bytes of one side of a flow, and how many bytes passed through.
@@ -797,3 +932,460 @@ async fn plain_http_request_looped_back_to_the_proxy_stops() {
     let error = find!(events, ProxyEvent::Error { error, .. } => error.clone());
     assert_eq!(error, "request looped back to Yaak Proxy");
 }
+
+// -- HTTP/2 clients --
+
+fn h2_request(method: &str, port: u16, path: &str, body: TestBody) -> Request<TestBody> {
+    Request::builder()
+        .method(method)
+        .uri(format!("https://localhost:{port}{path}"))
+        .body(body)
+        .unwrap()
+}
+
+/// The upstream test server, speaking HTTP/1.1 alone.
+fn h1_only(server: &ServerConfig) -> Arc<ServerConfig> {
+    let mut server = server.clone();
+    server.alpn_protocols = vec![b"http/1.1".to_vec()];
+    Arc::new(server)
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_client_to_an_h2_upstream() {
+    let h = Harness::start(1024);
+    let upstream = spawn_upstream(Some(h.pki.server.clone())).await;
+    let port = upstream.port();
+    let mut client = h.h2_client(port).await;
+
+    let resp = client.send_request(h2_request("GET", port, "/authority", full(""))).await.unwrap();
+    assert_eq!((resp.status(), resp.version()), (StatusCode::OK, Version::HTTP_2));
+    let text = resp.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(text, format!("authority=localhost:{port} host=none"));
+
+    let events = h.flow_events();
+    let (method, url, version, connection) = find!(events,
+        ProxyEvent::RequestStart { method, url, http_version, connection, .. } =>
+            (method.clone(), url.clone(), http_version.clone(), *connection));
+    assert_eq!(method, "GET");
+    assert_eq!(url, format!("https://localhost:{port}/authority"));
+    assert_eq!(version, "HTTP/2");
+    assert_eq!(connection, 1);
+    // `:authority` is shown as `Host`, once and first, as an HTTP/1.1 request has it.
+    let request_headers = headers(&events, true);
+    assert_eq!(request_headers.first(), Some(&("host".to_string(), format!("localhost:{port}"))));
+    assert_eq!(request_headers.iter().filter(|(name, _)| name == "host").count(), 1);
+
+    let (status, version, remote_addr) = find!(events,
+        ProxyEvent::ResponseStart { status, http_version, remote_addr, .. } =>
+            (*status, http_version.clone(), *remote_addr));
+    assert_eq!((status, version.as_str(), remote_addr), (200, "HTTP/2", Some(upstream)));
+    let (size, error, cancelled) = find!(events,
+        ProxyEvent::ResponseBodyComplete { size, error, cancelled, .. } =>
+            (*size, error.clone(), *cancelled));
+    assert_eq!((size, error, cancelled), (text.len() as u64, None, false));
+    assert_eq!(body(&events, false).0, text);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_client_to_an_h1_upstream() {
+    let h = Harness::start(1024);
+    let upstream = spawn_upstream(Some(h1_only(&h.pki.server))).await;
+    let port = upstream.port();
+    let mut client = h.h2_client(port).await;
+
+    let resp = client.send_request(h2_request("POST", port, "/echo", full("name=yaak"))).await;
+    let resp = resp.unwrap();
+    assert_eq!((resp.status(), resp.version()), (StatusCode::OK, Version::HTTP_2));
+    assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "name=yaak");
+    let events = h.flow_events();
+    let version = find!(events, ProxyEvent::RequestStart { http_version, .. } => http_version);
+    assert_eq!(version, "HTTP/2");
+    let (version, remote_addr) = find!(events,
+        ProxyEvent::ResponseStart { http_version, remote_addr, .. } =>
+            (http_version.clone(), *remote_addr));
+    assert_eq!((version.as_str(), remote_addr), ("HTTP/1.1", Some(upstream)));
+    assert_eq!(body(&events, true), (b"name=yaak".to_vec(), 9));
+
+    // `:authority` goes upstream as `Host`.
+    let resp = client.send_request(h2_request("GET", port, "/authority", full(""))).await;
+    let text = resp.unwrap().into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(text, format!("authority=none host=localhost:{port}"));
+    h.flow_events();
+
+    // Headers for the HTTP/1.1 connection alone are captured but not sent on to the client.
+    let resp = client.send_request(h2_request("GET", port, "/hop", full(""))).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    for name in ["connection", "x-hop", "keep-alive"] {
+        assert!(!resp.headers().contains_key(name), "{name} in {:?}", resp.headers());
+    }
+    assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "hop");
+    let captured = headers(&h.flow_events(), false);
+    assert!(captured.contains(&("x-hop".to_string(), "1".to_string())), "{captured:?}");
+}
+
+/// Each stream is a flow of its own, all on the client's one connection, and they share one
+/// connection upstream when the server speaks HTTP/2.
+#[tokio::test(flavor = "multi_thread")]
+async fn concurrent_h2_streams_are_flows_of_their_own() {
+    let h = Harness::start(1024);
+    let accepts = Arc::new(AtomicUsize::new(0));
+    let upstream = spawn_upstream_with(Some(h.pki.server.clone()), accepts.clone(), route).await;
+    let port = upstream.port();
+    let client = h.h2_client(port).await;
+
+    const STREAMS: usize = 20;
+    let sent = (0..STREAMS).map(|i| {
+        let mut client = client.clone();
+        async move {
+            let req = h2_request("POST", port, "/echo", full(format!("stream {i}")));
+            let resp = client.send_request(req).await.unwrap();
+            resp.into_body().collect().await.unwrap().to_bytes()
+        }
+    });
+    let bodies = futures_util::future::join_all(sent).await;
+    for (i, body) in bodies.iter().enumerate() {
+        assert_eq!(body, format!("stream {i}").as_bytes());
+    }
+
+    let flows = h.flows_events(STREAMS);
+    assert_eq!(flows.len(), STREAMS);
+    let connections: HashSet<u64> = flows
+        .values()
+        .map(|events| find!(events, ProxyEvent::RequestStart { connection, .. } => *connection))
+        .collect();
+    assert_eq!(connections.len(), 1, "one client connection");
+    let mut captured: Vec<String> =
+        flows.values().map(|events| String::from_utf8(body(events, true).0).unwrap()).collect();
+    captured.sort();
+    let mut expected: Vec<String> = (0..STREAMS).map(|i| format!("stream {i}")).collect();
+    expected.sort();
+    assert_eq!(captured, expected);
+    for events in flows.values() {
+        let echoed = body(events, false).0;
+        assert_eq!(echoed, body(events, true).0, "each flow keeps its own response");
+    }
+    assert_eq!(accepts.load(Ordering::SeqCst), 1, "the streams share one upstream connection");
+}
+
+/// An HTTP/1.1 server takes the streams on connections of its own, at once rather than one
+/// after another, and they are reused after.
+#[tokio::test(flavor = "multi_thread")]
+async fn concurrent_h2_streams_to_an_h1_upstream_run_at_once() {
+    const STREAMS: usize = 10;
+    let h = Harness::start(1024);
+    let accepts = Arc::new(AtomicUsize::new(0));
+    // Each `/together` request is held until all of them are in flight at once, which they can
+    // only be if none waits for another. Counted, not timed: how long a connection takes to
+    // open varies by machine.
+    let in_flight = Arc::new(AtomicUsize::new(0));
+    let most = Arc::new(AtomicUsize::new(0));
+    let handler = {
+        let (in_flight, most) = (in_flight.clone(), most.clone());
+        move |req: Request<Incoming>| {
+            let (in_flight, most) = (in_flight.clone(), most.clone());
+            async move {
+                if req.uri().path() != "/together" {
+                    return route(req).await;
+                }
+                most.fetch_max(in_flight.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
+                let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+                while in_flight.load(Ordering::SeqCst) < STREAMS
+                    && tokio::time::Instant::now() < deadline
+                {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+                Ok(Response::new(full("together")))
+            }
+        }
+    };
+    let upstream =
+        spawn_upstream_with(Some(h1_only(&h.pki.server)), accepts.clone(), handler).await;
+    let port = upstream.port();
+    let client = h.h2_client(port).await;
+
+    let sent = (0..STREAMS).map(|_| {
+        let mut client = client.clone();
+        async move {
+            let resp = client.send_request(h2_request("GET", port, "/together", full(""))).await;
+            resp.unwrap().into_body().collect().await.unwrap().to_bytes()
+        }
+    });
+    for body in futures_util::future::join_all(sent).await {
+        assert_eq!(body, "together");
+    }
+    assert_eq!(most.load(Ordering::SeqCst), STREAMS, "streams in flight upstream at once");
+    assert_eq!(h.flows_events(STREAMS).len(), STREAMS);
+    let opened = accepts.load(Ordering::SeqCst);
+    assert!((2..=STREAMS).contains(&opened), "{opened} upstream connections");
+
+    let mut client = client.clone();
+    let resp = client.send_request(h2_request("GET", port, "/hello", full(""))).await.unwrap();
+    assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "hello");
+    h.flow_events();
+    assert_eq!(accepts.load(Ordering::SeqCst), opened, "an idle connection is reused");
+}
+
+/// Sends its message when dropped, so a test sees the server give up on a request.
+struct OnDrop(tokio::sync::mpsc::UnboundedSender<&'static str>, &'static str);
+
+impl Drop for OnDrop {
+    fn drop(&mut self) {
+        let _ = self.0.send(self.1);
+    }
+}
+
+/// An upstream whose `/hang` never answers and whose `/stall` sends "hello" and then nothing,
+/// reporting on `tx` when each request arrives and when it is dropped.
+async fn spawn_stalling_upstream(
+    tls: Arc<ServerConfig>,
+    tx: tokio::sync::mpsc::UnboundedSender<&'static str>,
+) -> SocketAddr {
+    let handler = move |req: Request<Incoming>| {
+        let tx = tx.clone();
+        async move {
+            match req.uri().path() {
+                "/hang" => {
+                    let _guard = OnDrop(tx.clone(), "hang dropped");
+                    let _ = tx.send("hang arrived");
+                    match std::future::pending::<Infallible>().await {}
+                }
+                _ => {
+                    let guard = OnDrop(tx, "stall dropped");
+                    // The second frame never comes, and the guard goes with the body.
+                    let frames =
+                        futures_util::stream::unfold((true, guard), |(first, guard)| async move {
+                            if first {
+                                let _ = guard.0.send("stall sent");
+                                let hello = Ok(Frame::data(Bytes::from("hello")));
+                                return Some((hello, (false, guard)));
+                            }
+                            let _guard = guard;
+                            std::future::pending().await
+                        });
+                    Ok(Response::new(BodyExt::boxed(StreamBody::new(frames))))
+                }
+            }
+        }
+    };
+    spawn_upstream_with(Some(tls), Arc::new(AtomicUsize::new(0)), handler).await
+}
+
+async fn next_message(rx: &mut tokio::sync::mpsc::UnboundedReceiver<&'static str>) -> &'static str {
+    tokio::time::timeout(Duration::from_secs(10), rx.recv()).await.unwrap().unwrap()
+}
+
+/// A stream the client resets before the response is a cancelled flow, not an error, and the
+/// request upstream is cancelled with it.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_stream_cancelled_before_the_response() {
+    let h = Harness::start(1024);
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+    let upstream = spawn_stalling_upstream(h.pki.server.clone(), tx).await;
+    let port = upstream.port();
+    let mut client = h.h2_client(port).await;
+
+    let pending = tokio::spawn({
+        let mut client = client.clone();
+        async move { client.send_request(h2_request("GET", port, "/hang", full(""))).await }
+    });
+    assert_eq!(next_message(&mut rx).await, "hang arrived");
+    pending.abort();
+    assert_eq!(next_message(&mut rx).await, "hang dropped");
+
+    let events = h.flow_events();
+    find!(events, ProxyEvent::Cancelled { .. } => ());
+    assert!(!events.iter().any(|e| matches!(e, ProxyEvent::Error { .. })), "{events:#?}");
+
+    // The connection carries on.
+    let resp = client.send_request(h2_request("GET", port, "/stall", full(""))).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    assert_eq!(next_message(&mut rx).await, "stall sent");
+}
+
+/// A stream the client resets partway through the response body is cancelled too.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_stream_cancelled_during_the_body() {
+    let h = Harness::start(1024);
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+    let upstream = spawn_stalling_upstream(h.pki.server.clone(), tx).await;
+    let port = upstream.port();
+    let mut client = h.h2_client(port).await;
+
+    let resp = client.send_request(h2_request("GET", port, "/stall", full(""))).await.unwrap();
+    let mut body = resp.into_body();
+    let hello = body.frame().await.unwrap().unwrap().into_data().unwrap();
+    assert_eq!(hello, "hello");
+    assert_eq!(next_message(&mut rx).await, "stall sent");
+    drop(body);
+    assert_eq!(next_message(&mut rx).await, "stall dropped");
+
+    let events = h.flow_events();
+    let (size, error, cancelled) = find!(events,
+        ProxyEvent::ResponseBodyComplete { size, error, cancelled, .. } =>
+            (*size, error.clone(), *cancelled));
+    assert_eq!((size, error, cancelled), (5, None, true));
+}
+
+/// Trailers go both ways, as gRPC needs them to.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_trailers_pass_through() {
+    let h = Harness::start(1024);
+    for tls in [h.pki.server.clone(), h1_only(&h.pki.server)] {
+        let upstream = spawn_upstream(Some(tls)).await;
+        let port = upstream.port();
+        let mut client = h.h2_client(port).await;
+
+        let mut trailers = hyper::HeaderMap::new();
+        trailers.insert("x-client-trailer", HeaderValue::from_static("1"));
+        let frames = [
+            Ok(Frame::data(Bytes::from("message"))),
+            Ok(Frame::trailers(trailers)),
+        ];
+        let upload = BodyExt::boxed(StreamBody::new(futures_util::stream::iter(frames)));
+        let mut req = h2_request("POST", port, "/trailers", upload);
+        req.headers_mut().insert("te", HeaderValue::from_static("trailers"));
+        // HTTP/1.1 carries only trailers named up front, so an HTTP/1.1 server gets this one.
+        req.headers_mut().insert("trailer", HeaderValue::from_static("x-client-trailer"));
+
+        let resp = client.send_request(req).await.unwrap();
+        let collected = resp.into_body().collect().await.unwrap();
+        let trailers = collected.trailers().cloned().unwrap();
+        assert_eq!(trailers["x-client-trailer"], "1");
+        assert_eq!(trailers["x-upstream-trailer"], "yes");
+        assert_eq!(collected.to_bytes(), "message");
+
+        let events = h.flow_events();
+        let (size, error) = find!(events,
+            ProxyEvent::ResponseBodyComplete { size, error, .. } => (*size, error.clone()));
+        assert_eq!((size, error), (7, None));
+        assert_eq!(body(&events, true), (b"message".to_vec(), 7));
+    }
+}
+
+/// HTTP/2 may send a field per cookie; an HTTP/1.1 server gets them as the one `Cookie` it
+/// expects.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_cookies_are_joined_for_an_h1_upstream() {
+    let h = Harness::start(1024);
+    for (tls, expected) in [
+        (h1_only(&h.pki.server), "theme=dark; session=abc"),
+        (h.pki.server.clone(), "theme=dark | session=abc"),
+    ] {
+        let upstream = spawn_upstream(Some(tls)).await;
+        let port = upstream.port();
+        let mut client = h.h2_client(port).await;
+        let mut req = h2_request("GET", port, "/cookies", full(""));
+        req.headers_mut().append("cookie", HeaderValue::from_static("theme=dark"));
+        req.headers_mut().append("cookie", HeaderValue::from_static("session=abc"));
+        let resp = client.send_request(req).await.unwrap();
+        assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), expected);
+        h.flow_events();
+    }
+}
+
+/// A client that resets its stream partway through an upload cancels the flow, whichever
+/// version the server speaks.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_upload_cancelled_by_the_client() {
+    let h = Harness::start(1024);
+    for tls in [h1_only(&h.pki.server), h.pki.server.clone()] {
+        let upstream = spawn_upstream(Some(tls)).await;
+        let port = upstream.port();
+        let client = h.h2_client(port).await;
+
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Result<Frame<Bytes>, Infallible>>();
+        tx.send(Ok(Frame::data(Bytes::from("first part")))).unwrap();
+        let frames = futures_util::stream::unfold(rx, |mut rx| async move {
+            rx.recv().await.map(|frame| (frame, rx))
+        });
+        let upload = BodyExt::boxed(StreamBody::new(frames));
+        let pending = tokio::spawn({
+            let mut client = client.clone();
+            async move { client.send_request(h2_request("POST", port, "/echo", upload)).await }
+        });
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        pending.abort();
+        drop(tx);
+
+        let events = h.flow_events();
+        find!(events, ProxyEvent::Cancelled { .. } => ());
+        assert!(!events.iter().any(|e| matches!(e, ProxyEvent::Error { .. })), "{events:#?}");
+    }
+}
+
+/// An HTTP/2 request's body has no length of its own, which HTTP/1.1 needs, even on a GET.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_streamed_body_reaches_an_h1_upstream() {
+    let h = Harness::start(1024);
+    let upstream = spawn_upstream(Some(h1_only(&h.pki.server))).await;
+    let port = upstream.port();
+    let mut client = h.h2_client(port).await;
+
+    for method in ["GET", "POST"] {
+        let frames = ["streamed ", "body"].map(|s| Ok(Frame::data(Bytes::from(s))));
+        let upload = BodyExt::boxed(StreamBody::new(futures_util::stream::iter(frames)));
+        let resp = client.send_request(h2_request(method, port, "/echo", upload)).await.unwrap();
+        assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "streamed body");
+        let events = h.flow_events();
+        assert_eq!(body(&events, true), (b"streamed body".to_vec(), 13), "{method}");
+    }
+}
+
+/// More than the flow-control windows hold, both ways, through an HTTP/2 and an HTTP/1.1 server.
+#[tokio::test(flavor = "multi_thread")]
+async fn h2_large_bodies_pass_flow_control() {
+    const LARGE: usize = 12 * 1024 * 1024;
+    let h = Harness::start(1024);
+    for tls in [h.pki.server.clone(), h1_only(&h.pki.server)] {
+        let upstream = spawn_upstream(Some(tls)).await;
+        let port = upstream.port();
+        let mut client = h.h2_client(port).await;
+
+        let chunk = Bytes::from(vec![b'y'; BIG_BODY]);
+        let chunks = (0..LARGE / BIG_BODY).map(move |_| Ok(Frame::data(chunk.clone())));
+        let upload = BodyExt::boxed(StreamBody::new(futures_util::stream::iter(chunks)));
+        let resp = client.send_request(h2_request("POST", port, "/echo", upload)).await.unwrap();
+        let echoed = resp.into_body().collect().await.unwrap().to_bytes();
+        assert_eq!(echoed.len(), LARGE);
+        assert!(echoed.iter().all(|b| *b == b'y'));
+
+        let events = h.flow_events();
+        let (size, truncated) = find!(events,
+            ProxyEvent::RequestBodyComplete { size, truncated, .. } => (*size, *truncated));
+        assert_eq!((size, truncated), (LARGE as u64, true));
+        let (size, truncated, error) = find!(events,
+            ProxyEvent::ResponseBodyComplete { size, truncated, error, .. } =>
+                (*size, *truncated, error.clone()));
+        assert_eq!((size, truncated, error), (LARGE as u64, true, None));
+        assert_eq!(body(&events, false), (vec![b'y'; 1024], LARGE));
+    }
+}
+
+/// With HTTP/2 off, a client that offers both gets HTTP/1.1, from the next connection on.
+#[tokio::test(flavor = "multi_thread")]
+async fn http1_only_makes_h2_clients_fall_back() {
+    let h = Harness::start(1024);
+    let upstream = spawn_upstream(Some(h.pki.server.clone())).await;
+    let port = upstream.port();
+
+    h.proxy.set_http2(false);
+    let tls = h.tls_tunnel(port, &[b"h2", b"http/1.1"]).await;
+    assert_eq!(tls.get_ref().1.alpn_protocol(), Some(&b"http/1.1"[..]));
+    let (mut sender, conn) =
+        hyper::client::conn::http1::handshake(TokioIo::new(tls)).await.unwrap();
+    tokio::spawn(conn);
+    let req = Request::get("/hello").header(HOST, format!("localhost:{port}")).body(full(""));
+    let resp = sender.send_request(req.unwrap()).await.unwrap();
+    assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "hello");
+    let events = h.flow_events();
+    let version = find!(events, ProxyEvent::RequestStart { http_version, .. } => http_version);
+    assert_eq!(version, "HTTP/1.1");
+
+    h.proxy.set_http2(true);
+    let mut client = h.h2_client(port).await;
+    let resp = client.send_request(h2_request("GET", port, "/hello", full(""))).await.unwrap();
+    assert_eq!(resp.version(), Version::HTTP_2);
+    let events = h.flow_events();
+    let version = find!(events, ProxyEvent::RequestStart { http_version, .. } => http_version);
+    assert_eq!(version, "HTTP/2");
+}