From 1b133b6dbad8b7420ff7e68dfc5b85efca337d5f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 26 Sep 2026 01:48:15 +0000 Subject: [PATCH] fix(cdn): stop one CDN host hanging up from failing a download haru on Windows failed a wallpaper download with "fetch failed: malformed response: headers ended early": a CDN server accepted the TLS connection and closed it without answering. - Build the host list the way SteamKit and DepotDownloader do: drop ISP caches restricted to other apps, Steam China servers and proxies, and address each server by its vhost. Falls back to every reachable server if the filter would leave none. - Fetch manifests from up to four hosts, demoting the ones that fail, instead of asking one host once. - Throw away pooled connections idle for 15s, and make the HTTP retry dial a fresh connection rather than pop another pooled one. - Name the host in fetch errors. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01JPbjRcr2kuhRi1wBhGk1ax --- crates/tapline-cdn/src/lib.rs | 4 +- crates/tapline-rt-tokio/src/http.rs | 26 +++- crates/tapline/src/session.rs | 209 +++++++++++++++++++++------- 3 files changed, 181 insertions(+), 58 deletions(-) diff --git a/crates/tapline-cdn/src/lib.rs b/crates/tapline-cdn/src/lib.rs index 5899295..4b271a0 100644 --- a/crates/tapline-cdn/src/lib.rs +++ b/crates/tapline-cdn/src/lib.rs @@ -77,7 +77,7 @@ pub async fn fetch_chunk_bytes( let response = fetcher .get(Request::get(url), limit) .await - .map_err(|error| CdnError::Fetch(error.to_string()))?; + .map_err(|error| CdnError::Fetch(format!("{host}: {error}")))?; if !response.is_success() { return Err(CdnError::Status { @@ -161,7 +161,7 @@ pub async fn fetch_manifest( let response = fetcher .get(Request::get(url), tapline_manifest::MAX_MANIFEST as u64) .await - .map_err(|error| CdnError::Fetch(error.to_string()))?; + .map_err(|error| CdnError::Fetch(format!("{host}: {error}")))?; if !response.is_success() { return Err(CdnError::Status { diff --git a/crates/tapline-rt-tokio/src/http.rs b/crates/tapline-rt-tokio/src/http.rs index 529ecd7..0b659e3 100644 --- a/crates/tapline-rt-tokio/src/http.rs +++ b/crates/tapline-rt-tokio/src/http.rs @@ -1,14 +1,22 @@ use crate::tls::connect_tls; use std::collections::HashMap; use std::sync::Arc; +use std::time::{Duration, Instant}; use tapline_io::{Fetch, FetchError, Request, Response}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::Mutex; +/// How long a connection may sit in the pool before it is thrown away rather +/// than reused. CDN edges close idle keep-alive connections on their own +/// schedule, and a request written into one of those gets end-of-stream back +/// instead of a status line. +const IDLE_TIMEOUT: Duration = Duration::from_secs(15); + struct Connection { runtime: tokio::runtime::Id, stream: crate::tls::TlsStream, leftover: Vec, + idle_since: Instant, } pub struct HttpClient { @@ -33,13 +41,15 @@ impl HttpClient { } } - async fn acquire(&self, host: &str, port: u16) -> Result { + async fn acquire(&self, host: &str, port: u16, pooled: bool) -> Result { let here = tokio::runtime::Handle::try_current() .map_err(|error| FetchError::Transport(error.to_string()))? .id(); - if let Some(pool) = self.pools.lock().await.get_mut(host) { - pool.retain(|connection| connection.runtime == here); + if pooled && let Some(pool) = self.pools.lock().await.get_mut(host) { + pool.retain(|connection| { + connection.runtime == here && connection.idle_since.elapsed() < IDLE_TIMEOUT + }); if let Some(connection) = pool.pop() { return Ok(connection); } @@ -52,10 +62,12 @@ impl HttpClient { runtime: here, stream, leftover: Vec::new(), + idle_since: Instant::now(), }) } - async fn release(&self, host: &str, connection: Connection) { + async fn release(&self, host: &str, mut connection: Connection) { + connection.idle_since = Instant::now(); let mut pools = self.pools.lock().await; let idle: usize = pools.values().map(Vec::len).sum(); if idle >= self.idle_total { @@ -112,7 +124,11 @@ impl Fetch for HttpClient { let mut last_error = None; for attempt in 0..2 { - let connection = self.acquire(&host, port).await?; + // A pooled connection may have been closed by the server while it + // sat idle, which only shows once a request is written into it. + // The retry therefore always dials a new connection: popping a + // second pooled one could meet the same fate. + let connection = self.acquire(&host, port, attempt == 0).await?; match perform(connection, &host, &path, &request, limit).await { Ok((response, connection)) => { diff --git a/crates/tapline/src/session.rs b/crates/tapline/src/session.rs index 390920a..0958e65 100644 --- a/crates/tapline/src/session.rs +++ b/crates/tapline/src/session.rs @@ -20,7 +20,7 @@ use tapline_proto::steammessages_clientserver_2::{ }; use tapline_proto::steammessages_contentsystem_steamclient::{ CContentServerDirectory_GetManifestRequestCode_Request, - CContentServerDirectory_GetServersForSteamPipe_Request, + CContentServerDirectory_GetServersForSteamPipe_Request, CContentServerDirectory_ServerInfo, }; use tapline_proto::steammessages_publishedfile_steamclient::{ CPublishedFile_GetDetails_Request, CPublishedFile_Subscribe_Request, @@ -234,30 +234,56 @@ impl Session { }) .await?; - let hosts: Vec = directory - .servers - .iter() - .filter_map(|server| { - if !tapline_cdn::usable_over_tls(server.https_support.as_deref()) { - return None; - } - let host = server.host.clone()?; - Some(Host { - vhost: server.vhost.clone().unwrap_or_else(|| host.clone()), - host, - load: server - .load - .and_then(|value| u32::try_from(value).ok()) - .unwrap_or(u32::MAX), - https_required: server.https_support.as_deref() == Some("mandatory"), - }) - }) - .collect(); - + let hosts = cdn_hosts(&directory.servers); self.pool = HostPool::new(hosts); Ok(()) } + /// Fetch a manifest, moving on to another CDN host when one fails. A + /// single host that hangs up or errors must not fail the whole download, + /// which is what happened when this asked exactly one host once. + async fn fetch_manifest_from_pool( + &mut self, + depot: DepotId, + manifest_id: u64, + request_code: u64, + key: &[u8; 32], + ) -> Result { + const ATTEMPTS: usize = 4; + + let mut last_error = None; + for _ in 0..ATTEMPTS { + let host = match self.pool.acquire() { + Ok(host) => host, + Err(error) if last_error.is_none() => return Err(error.into()), + Err(_) => break, + }; + match fetch_manifest( + self.shared.http.as_ref(), + &host.host, + depot, + manifest_id, + request_code, + Some(key), + ) + .await + { + Ok(manifest) => { + self.pool.succeed(&host.host); + return Ok(manifest); + } + Err(error) => { + self.pool.demote(&host.host); + last_error = Some(error); + } + } + } + Err(last_error.map_or( + InstallError::Pool(tapline_cdn::PoolError::AllDemoted), + Into::into, + )) + } + pub async fn keep_alive(&mut self) -> Result<(), InstallError> { self.cm.heartbeat().await?; self.last_heartbeat = std::time::Instant::now(); @@ -349,16 +375,9 @@ impl Session { .manifest_request_code .unwrap_or(0); - let host = self.pool.acquire()?; - let manifest = fetch_manifest( - self.shared.http.as_ref(), - &host.host, - depot.id, - depot.manifest.get(), - code, - Some(&key), - ) - .await?; + let manifest = self + .fetch_manifest_from_pool(depot.id, depot.manifest.get(), code, &key) + .await?; resolved.push(ResolvedDepot { depot, @@ -421,16 +440,9 @@ impl Session { .manifest_request_code .unwrap_or(0); - let host = self.pool.acquire()?; - let content = fetch_manifest( - self.shared.http.as_ref(), - &host.host, - depot, - manifest.get(), - code, - Some(&key), - ) - .await?; + let content = self + .fetch_manifest_from_pool(depot, manifest.get(), code, &key) + .await?; Ok(ResolvedDepot { depot: tapline_pics::Depot { @@ -1188,16 +1200,9 @@ impl Session { .manifest_request_code .unwrap_or(0); - let host = self.pool.acquire()?; - let content = fetch_manifest( - self.shared.http.as_ref(), - &host.host, - *depot, - manifest.get(), - code, - Some(&key), - ) - .await?; + let content = self + .fetch_manifest_from_pool(*depot, manifest.get(), code, &key) + .await?; let entry = ResolvedDepot { depot: tapline_pics::Depot { @@ -1796,6 +1801,58 @@ fn now_unix() -> u64 { .unwrap_or(0) } +/// Turn Steam's content server directory into the hosts downloads may use. +/// +/// Steam hands back more than plain CDN edges: ISP caches that serve only the +/// apps in `allowed_app_ids`, Steam China servers, and proxies that expect a +/// rewritten request path. A request for anything else to one of those gets +/// its connection closed, which surfaced as "headers ended early". This keeps +/// the entries SteamKit and DepotDownloader use and, like them, addresses a +/// server by its `vhost`. If the filter would leave nothing, the unfiltered +/// list is kept so a directory shape this code does not know still works. +fn cdn_hosts(servers: &[CContentServerDirectory_ServerInfo]) -> Vec { + let reachable = |server: &&CContentServerDirectory_ServerInfo| { + tapline_cdn::usable_over_tls(server.https_support.as_deref()) + && server.host.as_deref().is_some_and(|host| !host.is_empty()) + }; + let general = |server: &&CContentServerDirectory_ServerInfo| { + server.allowed_app_ids.is_empty() + && server.steam_china_only != Some(true) + && server.use_as_proxy != Some(true) + && server + .r#type + .as_deref() + .is_none_or(|kind| matches!(kind, "CDN" | "SteamCache")) + }; + + let mut chosen: Vec<&CContentServerDirectory_ServerInfo> = + servers.iter().filter(reachable).filter(general).collect(); + if chosen.is_empty() { + chosen = servers.iter().filter(reachable).collect(); + } + + chosen + .into_iter() + .filter_map(|server| { + let host = server.host.clone()?; + let vhost = server + .vhost + .clone() + .filter(|vhost| !vhost.is_empty()) + .unwrap_or(host); + Some(Host { + host: vhost.clone(), + vhost, + load: server + .load + .and_then(|value| u32::try_from(value).ok()) + .unwrap_or(u32::MAX), + https_required: server.https_support.as_deref() == Some("mandatory"), + }) + }) + .collect() +} + #[cfg(test)] mod tests { use super::accounts_to_try; @@ -1860,4 +1917,54 @@ mod tests { crate::FileModes::SteamCmd ); } + + fn server(host: &str) -> super::CContentServerDirectory_ServerInfo { + super::CContentServerDirectory_ServerInfo { + r#type: Some("CDN".to_owned()), + host: Some(host.to_owned()), + vhost: Some(host.to_owned()), + load: Some(10), + https_support: Some("mandatory".to_owned()), + ..Default::default() + } + } + + #[test] + fn restricted_caches_china_servers_and_proxies_are_left_out() { + let mut isp = server("isp-cache.invalid"); + isp.r#type = Some("SteamCache".to_owned()); + isp.allowed_app_ids = vec![730]; + let mut china = server("china.invalid"); + china.steam_china_only = Some(true); + let mut proxy = server("proxy.invalid"); + proxy.use_as_proxy = Some(true); + let mut odd = server("open.invalid"); + odd.r#type = Some("OpenCache".to_owned()); + let mut no_tls = server("plain.invalid"); + no_tls.https_support = Some("unavailable".to_owned()); + + let hosts = super::cdn_hosts(&[isp, china, server("edge.invalid"), proxy, odd, no_tls]); + let names: Vec<&str> = hosts.iter().map(|host| host.host.as_str()).collect(); + assert_eq!(names, ["edge.invalid"]); + } + + #[test] + fn a_server_is_addressed_by_its_vhost() { + let mut edge = server("10-0-0-1.cache.invalid"); + edge.vhost = Some("cache1-ams1.steamcontent.invalid".to_owned()); + let mut bare = server("bare.invalid"); + bare.vhost = None; + + let hosts = super::cdn_hosts(&[edge, bare]); + assert_eq!(hosts[0].host, "cache1-ams1.steamcontent.invalid"); + assert_eq!(hosts[1].host, "bare.invalid"); + } + + #[test] + fn an_over_eager_filter_falls_back_to_every_reachable_server() { + let mut only = server("isp-cache.invalid"); + only.allowed_app_ids = vec![730]; + let hosts = super::cdn_hosts(&[only]); + assert_eq!(hosts.len(), 1); + } }