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); + } }