Stop one CDN host hanging up from failing a download

crates/tapline-cdn/src/lib.rs

@@ -77,7 +77,7 @@ pub async fn fetch_chunk_bytes<F: Fetch>(
     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<F: Fetch>(
     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 {

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<u8>,
+    idle_since: Instant,
 }
 
 pub struct HttpClient {
@@ -33,13 +41,15 @@ impl HttpClient {
         }
     }
 
-    async fn acquire(&self, host: &str, port: u16) -> Result<Connection, FetchError> {
+    async fn acquire(&self, host: &str, port: u16, pooled: bool) -> Result<Connection, FetchError> {
         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)) => {

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<Host> = 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<Manifest, InstallError> {
+        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<Host> {
+    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);
+    }
 }