From ebf2512d9689e671e1df9a6390c4009846804970 Mon Sep 17 00:00:00 2001 From: laoqijun Date: Sun, 8 Nov 2026 05:24:18 +0800 Subject: [PATCH 1/5] Steam: fix CDN downloads for international accounts on mainland-China networks The CM hands out only the Perfect World partner CDNs (dl.steam.clngaa.com, st.dl.eccdnx.com, xz.pphimalayanrt.com) to clients on mainland-Chinese networks. Those reject tokenless manifest/chunk requests from international accounts with HTTP 403, so no game could be downloaded without a proxy. Root causes and fixes: 1. GetCDNAuthToken protobuf field numbers were wrong (app_id/depot_id/host_name were 1/2/3; the protocol is depot_id/host_name/app_id). The CM returned an empty token, so the partner CDNs always 403'd. Fix the field order and thread the per-host token into manifest and chunk requests. 2. The logon response's cell_id was never saved, so GetServersForSteamPipe was called with cell_id=0. Persist it (matching the desktop client/SteamKit) and use it when requesting content servers. 3. The fallback CDN pool used cache*.steamcontent.com, which is behind DNS pollution and TLS SNI blocking on mainland networks. Fall back to the Asia-Pacific edges (dist*-hkg1 / dist*-sgp1) that are reachable from China and still honour international accounts. Also keep steam_china_only servers as a deprioritised fallback instead of discarding them, sweep every server on manifest/chunk retry, and emit per-host diagnostic logs. Verified on-device: Kaiju Princess 2 (3.21 GB) and Aokana (9.77 GB) download at 43-92 Mbps with no proxy. --- .../wn-steam-client/rust/src/cdn_client.rs | 26 +- .../cpp/wn-steam-client/rust/src/cm_client.rs | 31 ++ .../rust/src/depot_downloader.rs | 266 ++++++++++++++++-- .../wn-steam-client/rust/src/depot_writer.rs | 44 ++- .../main/cpp/wn-steam-client/rust/src/jni.rs | 68 ++++- .../src/pb/cmsg_client_get_cdn_auth_token.rs | 111 ++++++++ .../cpp/wn-steam-client/rust/src/pb/mod.rs | 1 + 7 files changed, 493 insertions(+), 54 deletions(-) create mode 100644 app/src/main/cpp/wn-steam-client/rust/src/pb/cmsg_client_get_cdn_auth_token.rs diff --git a/app/src/main/cpp/wn-steam-client/rust/src/cdn_client.rs b/app/src/main/cpp/wn-steam-client/rust/src/cdn_client.rs index 9c5bc9b56..efe134c21 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/cdn_client.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/cdn_client.rs @@ -292,7 +292,7 @@ impl CdnClient { if http_status != 200 { return CdnManifestResult { http_status, - error: "non-200 HTTP status".to_string(), + error: format!("non-200 HTTP status ({http_status})"), ..Default::default() }; } @@ -327,7 +327,7 @@ impl CdnClient { if http_status != 200 { return CdnChunkResult { http_status, - error: "non-200 HTTP status".to_string(), + error: format!("non-200 HTTP status ({http_status})"), ..Default::default() }; } @@ -388,7 +388,7 @@ impl CdnClient { .get(url) .timeout(timeout) .send() - .map_err(|err| format!("http get: {err}"))?; + .map_err(|err| format_http_error("http get", &err))?; let http_status = response.status().as_u16() as i32; let content_length = response.content_length(); let body = response @@ -440,6 +440,24 @@ struct HttpResponse { body: Vec, } +/// Expand a reqwest error into its full source chain so a TLS/connect failure +/// on a restricted network is diagnosable from a single log line (reqwest's +/// `Display` collapses everything into a terse "error sending request"). +fn format_http_error(context: &str, err: &reqwest::Error) -> String { + let mut out = format!("{context}: {err}"); + let mut source = std::error::Error::source(err); + let mut depth = 0; + while let Some(next) = source { + out.push_str(&format!(" | caused by: {next}")); + source = next.source(); + depth += 1; + if depth > 8 { + break; + } + } + out +} + impl CdnChunkResult { pub fn ok(&self) -> bool { self.error.is_empty() && !self.data.is_empty() @@ -636,7 +654,7 @@ mod tests { assert!(chunk.ok()); assert_eq!( CdnClient::validate_chunk_response(404, vec![], None).error, - "non-200 HTTP status" + "non-200 HTTP status (404)" ); } diff --git a/app/src/main/cpp/wn-steam-client/rust/src/cm_client.rs b/app/src/main/cpp/wn-steam-client/rust/src/cm_client.rs index b4e8d3fe5..488674028 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/cm_client.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/cm_client.rs @@ -142,6 +142,7 @@ pub struct CMClientCore { session_id: AtomicI32, family_group_id: AtomicU64, server_realtime: AtomicU32, + cell_id: AtomicU32, playing_blocked: AtomicBool, license_list: Mutex>, friends: Mutex>, @@ -170,6 +171,7 @@ impl Default for CMClientCore { session_id: AtomicI32::new(0), family_group_id: AtomicU64::new(0), server_realtime: AtomicU32::new(0), + cell_id: AtomicU32::new(0), playing_blocked: AtomicBool::new(false), license_list: Mutex::new(Vec::new()), friends: Mutex::new(HashMap::new()), @@ -202,6 +204,7 @@ impl CMClientCore { self.session_id.store(0, Ordering::Relaxed); self.family_group_id.store(0, Ordering::Relaxed); self.server_realtime.store(0, Ordering::Relaxed); + self.cell_id.store(0, Ordering::Relaxed); self.outbound_wires .lock() .expect("outbound queue poisoned") @@ -224,6 +227,10 @@ impl CMClientCore { self.server_realtime.load(Ordering::Relaxed) } + pub fn cell_id(&self) -> u32 { + self.cell_id.load(Ordering::Relaxed) + } + pub fn is_playing_blocked(&self) -> bool { self.playing_blocked.load(Ordering::Relaxed) } @@ -724,6 +731,28 @@ impl CMClientCore { ) } + pub fn build_get_cdn_auth_token( + &self, + app_id: u32, + depot_id: u32, + host_name: &str, + job_id: u64, + ) -> Option { + if app_id == 0 || depot_id == 0 || host_name.is_empty() { + return None; + } + self.build_authed_service_call( + "ContentServerDirectory.GetCDNAuthToken#1", + job_id, + crate::pb::cmsg_client_get_cdn_auth_token::CMsgClientGetCdnAuthToken { + app_id, + depot_id, + host_name: host_name.to_string(), + } + .serialize(), + ) + } + pub fn build_manifest_request_code_call( &self, app_id: u32, @@ -1359,6 +1388,7 @@ impl CMClientCore { .store(resp.family_group_id, Ordering::Relaxed); self.server_realtime .store(resp.rtime32_server_time, Ordering::Relaxed); + self.cell_id.store(resp.cell_id, Ordering::Relaxed); InboundAction::LogonOk } else { InboundAction::LoggedOff @@ -1373,6 +1403,7 @@ impl CMClientCore { self.session_id.store(0, Ordering::Relaxed); self.family_group_id.store(0, Ordering::Relaxed); self.server_realtime.store(0, Ordering::Relaxed); + self.cell_id.store(0, Ordering::Relaxed); InboundAction::LoggedOff } EMsg::CLIENT_LICENSE_LIST => { diff --git a/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs b/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs index 3bd132a0c..bdf9e098e 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs @@ -9,8 +9,6 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; use std::time::Duration; -pub const MAX_MANIFEST_FETCH_ATTEMPTS: usize = 5; - #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct DepotSpec { pub depot_id: u32, @@ -137,22 +135,88 @@ pub fn validate_resolved_download_inputs( Ok(()) } +/// Keeps every CDN server that has a reachable host (or vhost) while ordering +/// the list so that globally-reachable servers are tried first and China-only +/// servers act as a last-resort fallback. +/// +/// Previously `steam_china_only` servers were discarded outright. On networks +/// where the international `*.steamcontent.com` CDN is unreachable or +/// unstable (e.g. mainland China), that left the downloader with no usable +/// server at all, so manifest/chunk fetches failed before a single byte could +/// be downloaded. Keeping them — but deprioritising them — preserves the fast +/// path for the rest of the world while giving those networks a working path. pub fn filter_usable_cdn_servers( servers: impl IntoIterator, ) -> Vec { - servers - .into_iter() - .filter(|server| !server.steam_china_only && !server.host.is_empty()) + let mut global = Vec::new(); + let mut china = Vec::new(); + for server in servers { + // A server is usable if either `host` or `vhost` is present; the CDN + // client prefers `vhost` for the URL/SNI and falls back to `host`. + let has_host = !server.host.is_empty() || !server.vhost.is_empty(); + if !has_host { + continue; + } + if server.steam_china_only { + china.push(server); + } else { + global.push(server); + } + } + global.extend(china); + // The CM picks CDN nodes by the device's network location. On a device + // inside mainland China it hands out only the Perfect World partner CDNs + // (dl.steam.clngaa.com / st.dl.eccdnx.com / xz.pphimalayanrt.com), which + // reject manifests for non-PerfectWorld (international) accounts with + // HTTP 403 — leaving the downloader with zero working nodes. Append the + // global Valve CDN pool as a last-resort fallback so international + // accounts on Chinese networks can still fetch manifests and chunks. + global.extend(global_cdn_fallback_servers()); + global +} + +/// Global Valve CDN hosts that serve any depot regardless of account region. +/// +/// These are Steam's Asia-Pacific edge nodes (`dist*-hkg1` / `dist*-sgp1`). +/// Unlike the `cache*.steamcontent.com` pool — which is behind DNS pollution +/// and TLS SNI blocking on mainland-Chinese networks — the Hong Kong and +/// Singapore edges are served from Akamai addresses that are directly +/// reachable from China while still honouring international Steam accounts +/// (a 401 "needs a valid manifest request code" instead of the 403 that the +/// Perfect World partner CDNs return for international accounts). +fn global_cdn_fallback_servers() -> Vec { + const GLOBAL_CDN_HOSTS: &[&str] = &[ + "dist1-hkg1.steamcontent.com", + "dist2-hkg1.steamcontent.com", + "dist3-hkg1.steamcontent.com", + "dist1-sgp1.steamcontent.com", + "dist2-sgp1.steamcontent.com", + "dist3-sgp1.steamcontent.com", + // Last-resort global edges in case the Asia-Pacific pool is unreachable. + "cache1.steamcontent.com", + "cache2.steamcontent.com", + ]; + GLOBAL_CDN_HOSTS + .iter() + .map(|host| CContentServerDirectoryServerInfo { + server_type: "CDN".to_string(), + host: (*host).to_string(), + vhost: (*host).to_string(), + https_support: "mandatory".to_string(), + ..Default::default() + }) .collect() } -pub fn manifest_retry_server_indices(server_count: usize, attempts: usize) -> Vec { +pub fn manifest_retry_server_indices(server_count: usize) -> Vec { if server_count == 0 { return Vec::new(); } - (0..attempts) - .map(|attempt| attempt % server_count) - .collect() + // Try each server in order exactly once. When the list mixes global and + // China-only fallback servers, a fixed round-robin window (e.g. 5 attempts) + // could exhaust itself on the global servers and never reach the fallback + // that actually works on a restricted network. + (0..server_count).collect() } pub fn retry_backoff_millis(attempt: u32) -> u64 { @@ -163,6 +227,10 @@ pub fn retry_backoff_millis(attempt: u32) -> u64 { } } +/// Returns the CDN auth token for a specific depot + host. Region-restricted +/// CDN pools require a per-host token to serve manifests/chunks. +pub type CdnAuthTokenRefresher<'a> = &'a (dyn Fn(u32, &str) -> Option + Sync); + pub fn fetch_manifest_with_retry( cdn: &CdnClient, servers: &[CContentServerDirectoryServerInfo], @@ -171,6 +239,28 @@ pub fn fetch_manifest_with_retry( request_code: u64, cdn_auth_token: &str, timeout: Duration, +) -> CdnManifestResult { + fetch_manifest_with_retry_and_token( + cdn, + servers, + depot_id, + manifest_id, + request_code, + cdn_auth_token, + None, + timeout, + ) +} + +pub fn fetch_manifest_with_retry_and_token( + cdn: &CdnClient, + servers: &[CContentServerDirectoryServerInfo], + depot_id: u32, + manifest_id: u64, + request_code: u64, + fallback_token: &str, + token_refresher: Option>, + timeout: Duration, ) -> CdnManifestResult { if servers.is_empty() { return CdnManifestResult { @@ -178,30 +268,95 @@ pub fn fetch_manifest_with_retry( ..Default::default() }; } + // Sweep every server once (global first, then China-only fallback) so a + // restricted network can still reach a working CDN node. A fixed retry + // window would otherwise burn itself on the first few global servers and + // never touch the fallback list. let mut last = CdnManifestResult::default(); + let mut seen_errors: Vec = Vec::new(); for (attempt, server_idx) in - manifest_retry_server_indices(servers.len(), MAX_MANIFEST_FETCH_ATTEMPTS) + manifest_retry_server_indices(servers.len()) .into_iter() .enumerate() { if attempt > 0 { thread::sleep(Duration::from_millis(retry_backoff_millis(attempt as u32))); } + let server = &servers[server_idx]; + // A region-restricted host rejects tokenless requests with 403; fetch a + // per-host token on demand when the caller provided a provider. + let token = token_refresher + .and_then(|refresh| refresh(depot_id, preferred_cdn_host(server))) + .unwrap_or_else(|| fallback_token.to_string()); last = cdn.fetch_manifest( - &servers[server_idx], + server, depot_id, manifest_id, request_code, - cdn_auth_token, + &token, timeout, ); if last.ok() { return last; } + // Record one line per distinct failure so a remote-diagnosis log shows + // which CDN hosts were tried and how each rejected the request. + let summary = format!( + "manifest depot {depot_id} via {}:{} -> {}", + preferred_cdn_host(server), + last.http_status, + last.error + ); + if !seen_errors.contains(&summary) { + seen_errors.push(summary.clone()); + log_line(&summary); + } + } + if !seen_errors.is_empty() { + log_line(&format!( + "manifest depot {depot_id} failed on all {} servers; last: {}", + servers.len(), + last.error + )); } last } +fn preferred_cdn_host(server: &CContentServerDirectoryServerInfo) -> &str { + if !server.vhost.is_empty() { + &server.vhost + } else { + &server.host + } +} + +// One-line Android log emit; kept dependency-free so the downloader stays +// testable off-device (no android linkage required for unit tests). +pub fn log_line(message: &str) { + #[cfg(target_os = "android")] + { + use std::ffi::CString; + let tag = CString::new("WnDepotDownloader").unwrap_or_default(); + let msg = CString::new(message).unwrap_or_default(); + unsafe { + // __android_log_print(ANDROID_LOG_INFO, tag, fmt, ...) + extern "C" { + fn __android_log_print( + prio: i32, + tag: *const std::os::raw::c_char, + fmt: *const std::os::raw::c_char, + ... + ) -> i32; + } + let _ = __android_log_print(4, tag.as_ptr(), b"%s\0".as_ptr() as *const _, msg.as_ptr()); + } + } + #[cfg(not(target_os = "android"))] + { + eprintln!("{message}"); + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum DepotResumeDecision { SkipInstalled, @@ -268,6 +423,8 @@ pub fn download_resolved_depots( None, None, None, + "", + None, ) } @@ -290,6 +447,8 @@ pub fn download_resolved_depots_with_cancel( cancel, None, None, + "", + None, ) } @@ -303,6 +462,8 @@ pub fn download_resolved_depots_with_cancel_progress( cancel: Option<&AtomicBool>, on_progress: Option>, code_refresher: Option>, + cdn_auth_token: &str, + token_refresher: Option>, ) -> DepotDownloadResult { if let Err(error) = validate_resolved_download_inputs(install_dir, depots, servers) { return error; @@ -375,13 +536,14 @@ pub fn download_resolved_depots_with_cancel_progress( let refreshed_code = code_refresher .and_then(|refresh| refresh(depot.depot_id, depot.manifest_id)); let request_code = refreshed_code.unwrap_or(depot.manifest_request_code); - let mut manifest = fetch_manifest_with_retry( + let mut manifest = fetch_manifest_with_retry_and_token( &cdn, &usable_servers, depot.depot_id, depot.manifest_id, request_code, - "", + cdn_auth_token, + token_refresher, CdnClient::default_timeout(), ); if !manifest.ok() { @@ -390,13 +552,14 @@ pub fn download_resolved_depots_with_cancel_progress( .and_then(|refresh| refresh(depot.depot_id, depot.manifest_id)) .filter(|fresh| *fresh != request_code) { - manifest = fetch_manifest_with_retry( + manifest = fetch_manifest_with_retry_and_token( &cdn, &usable_servers, depot.depot_id, depot.manifest_id, fresh, - "", + cdn_auth_token, + token_refresher, CdnClient::default_timeout(), ); } @@ -452,6 +615,7 @@ pub fn download_resolved_depots_with_cancel_progress( max_workers, cancel, on_progress: Some(chunk_progress), + cdn_auth_token, ..Default::default() }, ); @@ -523,24 +687,75 @@ mod tests { } #[test] - fn filters_non_china_servers_with_hosts() { + fn keeps_china_servers_as_fallback_and_drops_hostless() { let servers = filter_usable_cdn_servers([ CContentServerDirectoryServerInfo { - host: "ok".into(), + host: "global1".into(), ..Default::default() }, CContentServerDirectoryServerInfo { - host: "china".into(), + host: "china1".into(), steam_china_only: true, ..Default::default() }, + CContentServerDirectoryServerInfo { + host: "global2".into(), + ..Default::default() + }, + // vhost-only servers are still reachable (CDN client falls back to vhost). + CContentServerDirectoryServerInfo { + host: String::new(), + vhost: "cdn-vhost.example".into(), + ..Default::default() + }, + // No host and no vhost: unusable, must be dropped. CContentServerDirectoryServerInfo { host: String::new(), ..Default::default() }, + CContentServerDirectoryServerInfo { + host: "china2".into(), + steam_china_only: true, + ..Default::default() + }, + ]); + // Global servers first, then China-only, then the global Valve CDN + // fallback pool; hostless entries dropped. + assert_eq!(servers.len(), 5 + 8); + assert_eq!(servers[0].host, "global1"); + assert_eq!(servers[1].host, "global2"); + assert_eq!(servers[2].vhost, "cdn-vhost.example"); + assert_eq!(servers[3].host, "china1"); + assert!(servers[3].steam_china_only); + assert_eq!(servers[4].host, "china2"); + assert!(servers[4].steam_china_only); + // The tail is the Valve global CDN fallback pool (Asia-Pacific edges first). + assert_eq!(servers[5].host, "dist1-hkg1.steamcontent.com"); + assert!(servers.last().unwrap().host.ends_with("cache2.steamcontent.com")); + } + + #[test] + fn appends_global_fallback_when_only_partner_cdns_offered() { + // Reproduces the field failure: a CM on a mainland-Chinese network + // hands out only Perfect World partner CDNs (china_only=false!), which + // 403 international accounts. The global pool must be appended. + let servers = filter_usable_cdn_servers([ + CContentServerDirectoryServerInfo { + host: "st.dl.eccdnx.com".into(), + vhost: "st.dl.eccdnx.com".into(), + ..Default::default() + }, + CContentServerDirectoryServerInfo { + host: "dl.steam.clngaa.com".into(), + vhost: "dl.steam.clngaa.com".into(), + ..Default::default() + }, ]); - assert_eq!(servers.len(), 1); - assert_eq!(servers[0].host, "ok"); + assert_eq!(servers.len(), 2 + 8); + assert_eq!(servers[0].host, "st.dl.eccdnx.com"); + assert_eq!(servers[1].host, "dl.steam.clngaa.com"); + assert_eq!(servers[2].host, "dist1-hkg1.steamcontent.com"); + assert_eq!(servers[2].port(), 443); } #[test] @@ -580,9 +795,12 @@ mod tests { } #[test] - fn retry_rotation_and_progress_mapping_match_cpp() { - assert_eq!(manifest_retry_server_indices(3, 5), [0, 1, 2, 0, 1]); - assert_eq!(manifest_retry_server_indices(0, 5), Vec::::new()); + fn retry_sweep_and_progress_mapping() { + // The manifest retry sweeps each server once (in order), so fallback + // servers at the tail are still reached. + assert_eq!(manifest_retry_server_indices(3), [0, 1, 2]); + assert_eq!(manifest_retry_server_indices(1), [0]); + assert_eq!(manifest_retry_server_indices(0), Vec::::new()); assert_eq!(retry_backoff_millis(1), 300); assert_eq!(retry_backoff_millis(5), 4000); assert_eq!( diff --git a/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs b/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs index 16e9b490d..84258918b 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs @@ -13,7 +13,6 @@ use std::time::{Duration, Instant}; pub const DEPOT_FILE_FLAG_EXECUTABLE: u32 = 32; pub const DEPOT_FILE_FLAG_DIRECTORY: u32 = 64; pub const DEPOT_FILE_FLAG_SYMLINK: u32 = 512; -pub const MAX_CHUNK_ATTEMPTS: u32 = 5; const RESIZE_BUFFER_BYTES: usize = 256 * 1024; const ERRNO_EPERM: i32 = 1; const ERRNO_EACCES: i32 = 13; @@ -108,18 +107,18 @@ pub fn retry_backoff_millis(attempt: u32) -> u64 { pub fn chunk_attempt_server_indices( start_server_index: usize, server_count: usize, - attempts: u32, ) -> Vec { - if server_count == 0 || attempts == 0 { + if server_count == 0 { return Vec::new(); } - let mut out = Vec::with_capacity(attempts as usize); - let mut server_index = start_server_index % server_count; - for attempt in 0..attempts { - if attempt > 0 && server_count > 1 { - server_index = (server_index + 1) % server_count; - } - out.push(server_index); + // Sweep every server exactly once, starting from `start_server_index` and + // wrapping around. This guarantees the China-only fallback servers (kept at + // the tail of the usable-server list) are actually reached when the global + // CDN nodes fail fast (connection reset / timeout on restricted networks). + let mut out = Vec::with_capacity(server_count); + let start = start_server_index % server_count; + for offset in 0..server_count { + out.push((start + offset) % server_count); } out } @@ -578,11 +577,8 @@ pub fn fetch_process_write_chunk( .ok_or_else(|| "write_depot: bad chunk index".to_string())?; let path = join_target_path(target_dir, &file.filename); let mut last_error = String::new(); - for (attempt, server_idx) in - chunk_attempt_server_indices(start_server_index, servers.len(), MAX_CHUNK_ATTEMPTS) - .into_iter() - .enumerate() - { + let attempt_indices = chunk_attempt_server_indices(start_server_index, servers.len()); + for (attempt, server_idx) in attempt_indices.into_iter().enumerate() { if attempt > 0 { thread::sleep(Duration::from_millis(retry_backoff_millis(attempt as u32))); if let Some(connection) = conn.as_deref_mut() { @@ -616,8 +612,10 @@ pub fn fetch_process_write_chunk( } } Err(format!( - "write_depot: chunk for '{}' failed after {} attempts: {}", - file.filename, MAX_CHUNK_ATTEMPTS, last_error + "write_depot: chunk for '{}' failed after trying all {} servers: {}", + file.filename, + servers.len(), + last_error )) } @@ -925,16 +923,16 @@ mod tests { assert_eq!(retry_backoff_millis(1), 300); assert_eq!(retry_backoff_millis(2), 600); assert_eq!(retry_backoff_millis(5), 4000); - assert_eq!(MAX_CHUNK_ATTEMPTS, 5); assert_eq!(SLOW_CHUNK_ROTATE_THRESHOLD_SECS, 8); } #[test] - fn chunk_retry_rotates_across_servers_like_cpp_worker() { - assert_eq!(chunk_attempt_server_indices(0, 3, 5), [0, 1, 2, 0, 1]); - assert_eq!(chunk_attempt_server_indices(2, 3, 5), [2, 0, 1, 2, 0]); - assert_eq!(chunk_attempt_server_indices(0, 1, 5), [0, 0, 0, 0, 0]); - assert!(chunk_attempt_server_indices(0, 0, 5).is_empty()); + fn chunk_retry_sweeps_every_server_starting_from_bias() { + assert_eq!(chunk_attempt_server_indices(0, 3), [0, 1, 2]); + assert_eq!(chunk_attempt_server_indices(2, 3), [2, 0, 1]); + assert_eq!(chunk_attempt_server_indices(1, 1), [0]); + assert_eq!(chunk_attempt_server_indices(5, 3), [2, 0, 1]); + assert!(chunk_attempt_server_indices(0, 0).is_empty()); assert!(!should_rotate_after_slow_chunks(2, 3)); assert!(should_rotate_after_slow_chunks(3, 3)); assert!(!should_rotate_after_slow_chunks(3, 1)); diff --git a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs index e2e470140..04cbf3593 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs @@ -1241,11 +1241,60 @@ fn request_cdn_servers( runtime: &Arc, timeout: Duration, ) -> Option> { + // Use the cell_id the CM handed back at logon (like the desktop client / + // SteamKit does) rather than 0. A concrete cell_id makes the directory + // service return the correct regional edge pool instead of a fallback + // China-only partner list. + let cell_id = runtime.core().cell_id(); let body = request_authed_service_body(runtime, timeout, |core, job_id| { - core.build_get_cdn_servers_call(0, job_id) + core.build_get_cdn_servers_call(cell_id, job_id) })?; - crate::pb::ccontentserverdirectory::CContentServerDirectoryGetServersForSteamPipeResponse::deserialize(&body) - .map(|response| response.servers) + let servers = crate::pb::ccontentserverdirectory::CContentServerDirectoryGetServersForSteamPipeResponse::deserialize(&body) + .map(|response| response.servers)?; + // Diagnose restricted networks: log every node the CM offered, including + // its china-only flag, so a 403 from the wrong CDN pool is explainable + // from logcat alone. + for server in &servers { + crate::depot_downloader::log_line(&format!( + "CM cdn server: host={} vhost={} type={} china_only={} https={} allowed_apps={}", + server.host, + server.vhost, + server.server_type, + server.steam_china_only, + server.https_support, + server.allowed_app_ids.len(), + )); + } + Some(servers) +} + +fn request_cdn_auth_token( + runtime: &Arc, + app_id: u32, + depot_id: u32, + host_name: &str, + timeout: Duration, +) -> Option { + crate::depot_downloader::log_line(&format!( + "requesting CDN auth token for app {app_id} depot {depot_id} host {host_name}" + )); + let body = request_authed_service_body(runtime, timeout, |core, job_id| { + core.build_get_cdn_auth_token(app_id, depot_id, host_name, job_id) + })?; + let response = + crate::pb::cmsg_client_get_cdn_auth_token::CMsgClientGetCdnAuthTokenResponse::deserialize(&body)?; + if !response.token.is_empty() { + crate::depot_downloader::log_line(&format!( + "CDN auth token acquired for host {host_name} ({} chars)", + response.token.len() + )); + Some(response.token) + } else { + crate::depot_downloader::log_line(&format!( + "CDN auth token empty for host {host_name}" + )); + None + } } const ERESULT_OK: i32 = 1; @@ -3327,6 +3376,17 @@ pub extern "system" fn Java_com_winlator_cmod_feature_stores_steam_wnsteam_WnSte ); return; }; + // Region-restricted CDN pools (e.g. the Perfect World partner CDNs the + // CM hands out on mainland-Chinese networks) reject tokenless + // manifest/chunk requests with HTTP 403. Provide a per-host token + // refresher that the downloader invokes lazily; on failure it falls + // back to an empty token, which the global Valve CDN pool still honours. + let cdn_auth_token = String::new(); + let token_refresher = |depot_id: u32, host_name: &str| -> Option { + request_cdn_auth_token(&runtime, app_id, depot_id, host_name, timeout) + }; + let token_refresher_cb: crate::depot_downloader::CdnAuthTokenRefresher = + &token_refresher; let config_dir = std::path::Path::new(&install_dir).join(".DepotDownloader"); let installed_cfg = crate::depot_config::DepotConfigStore::load(&config_dir); let mut resolved = Vec::with_capacity(specs.len()); @@ -3463,6 +3523,8 @@ pub extern "system" fn Java_com_winlator_cmod_feature_stores_steam_wnsteam_WnSte Some(download_cancel.as_ref()), Some(progress_cb), Some(code_refresher_cb), + &cdn_auth_token, + Some(token_refresher_cb), ); dispatch_download_complete(listener, result); }); diff --git a/app/src/main/cpp/wn-steam-client/rust/src/pb/cmsg_client_get_cdn_auth_token.rs b/app/src/main/cpp/wn-steam-client/rust/src/pb/cmsg_client_get_cdn_auth_token.rs new file mode 100644 index 000000000..0a491ffa1 --- /dev/null +++ b/app/src/main/cpp/wn-steam-client/rust/src/pb/cmsg_client_get_cdn_auth_token.rs @@ -0,0 +1,111 @@ +use crate::proto_wire::{Reader, Writer}; + +/// CContentServerDirectory.GetCDNAuthToken#1 request (SteamKit unified +/// service): app/depot/host triple. Region-restricted CDN pools (e.g. the +/// Perfect World partner CDNs inside mainland China) reject tokenless +/// manifest/chunk requests with HTTP 403, so the downloader must fetch this. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct CMsgClientGetCdnAuthToken { + pub app_id: u32, + pub depot_id: u32, + pub host_name: String, +} + +impl CMsgClientGetCdnAuthToken { + pub fn serialize(&self) -> Vec { + let mut out = Vec::new(); + let mut w = Writer::new(&mut out); + // Field numbers per SteamKit's CContentServerDirectory_GetCDNAuthToken_Request: + // 1 = depot_id, 2 = host_name, 3 = app_id. + w.uint32_field(1, self.depot_id); + w.string_field(2, &self.host_name); + w.uint32_field(3, self.app_id); + out + } +} + +/// GetCDNAuthToken#1 response: the CDN auth token plus its expiry. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct CMsgClientGetCdnAuthTokenResponse { + pub token: String, + pub expiration_time: u32, +} + +impl CMsgClientGetCdnAuthTokenResponse { + pub fn deserialize(body: &[u8]) -> Option { + let mut reader = Reader::new(body); + let mut msg = Self::default(); + while !reader.eof() { + let Some(tag) = reader.next_tag() else { + return reader.ok().then_some(msg); + }; + match tag.field_number { + 1 => msg.token = reader.string()?, + 2 => msg.expiration_time = reader.u32()?, + _ => { + if !reader.skip(tag.wire_type) { + return None; + } + } + } + } + Some(msg) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn request_serializes_all_fields() { + let req = CMsgClientGetCdnAuthToken { + app_id: 2289720, + depot_id: 2289721, + host_name: "st.dl.eccdnx.com".into(), + }; + let bytes = req.serialize(); + // Round-trip via manual parse: field1 = depot_id (varint), + // field2 = host_name (string), field3 = app_id (varint). + let mut r = Reader::new(&bytes); + let mut depot = None; + let mut host = None; + let mut app = None; + while let Some(tag) = r.next_tag() { + match tag.field_number { + 1 => depot = Some(r.u32().unwrap()), + 2 => host = Some(r.string().unwrap()), + 3 => app = Some(r.u32().unwrap()), + _ => panic!("unexpected field"), + } + } + assert_eq!(depot, Some(2289721)); + assert_eq!(host.as_deref(), Some("st.dl.eccdnx.com")); + assert_eq!(app, Some(2289720)); + } + + #[test] + fn response_parses_token_fields() { + let mut body = Vec::new(); + { + let mut w = Writer::new(&mut body); + w.string_field(1, "cdn-token-abc"); + w.uint32_field(2, 1769000000); + } + let parsed = CMsgClientGetCdnAuthTokenResponse::deserialize(&body).unwrap(); + assert_eq!(parsed.token, "cdn-token-abc"); + assert_eq!(parsed.expiration_time, 1769000000); + } + + #[test] + fn response_skips_unknown_fields() { + let mut body = Vec::new(); + { + let mut w = Writer::new(&mut body); + w.uint32_field(9, 3); + w.string_field(1, "t"); + } + let parsed = CMsgClientGetCdnAuthTokenResponse::deserialize(&body).unwrap(); + assert_eq!(parsed.token, "t"); + } +} diff --git a/app/src/main/cpp/wn-steam-client/rust/src/pb/mod.rs b/app/src/main/cpp/wn-steam-client/rust/src/pb/mod.rs index 27a82e1b4..68709e5a8 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/pb/mod.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/pb/mod.rs @@ -8,6 +8,7 @@ pub mod cmsg_client_change_status; pub mod cmsg_client_friends_list; pub mod cmsg_client_games_played; pub mod cmsg_client_get_app_ownership_ticket; +pub mod cmsg_client_get_cdn_auth_token; pub mod cmsg_client_get_depot_decryption_key; pub mod cmsg_client_get_user_stats; pub mod cmsg_client_kick_playing_session; From f2ee048fcc8c0765a2a3d71835058dfaeb018d8a Mon Sep 17 00:00:00 2001 From: laoqijun Date: Sun, 8 Nov 2026 05:54:42 +0800 Subject: [PATCH 2/5] Stores: add Turbo (48) and Maximum (64) download concurrency tiers The download speed setting capped concurrency at 32 workers. Measurement on a mainland-China network showed the app using only ~1.2 cores of CPU while downloading, i.e. the bottleneck is per-connection CDN throughput rather than local processing, so exposing higher worker counts lets fast devices and connections pull more bandwidth in parallel. Adds two tiers above Performance: Turbo (48) and Maximum (64). The native downloader already clamps to 64. --- app/src/main/feature/settings/stores/StoresScreen.kt | 2 ++ app/src/main/res/values/strings.xml | 2 ++ 2 files changed, 4 insertions(+) diff --git a/app/src/main/feature/settings/stores/StoresScreen.kt b/app/src/main/feature/settings/stores/StoresScreen.kt index caac4b8dc..dedcf0da8 100644 --- a/app/src/main/feature/settings/stores/StoresScreen.kt +++ b/app/src/main/feature/settings/stores/StoresScreen.kt @@ -188,6 +188,8 @@ fun StoresScreen( 16 to stringResource(R.string.stores_accounts_download_speed_balanced), 24 to stringResource(R.string.stores_accounts_download_speed_standard), 32 to stringResource(R.string.stores_accounts_download_speed_performance), + 48 to stringResource(R.string.stores_accounts_download_speed_turbo), + 64 to stringResource(R.string.stores_accounts_download_speed_maximum), ) CompositionLocalProvider(LocalPaneNav provides contentNav) { diff --git a/app/src/main/res/values/strings.xml b/app/src/main/res/values/strings.xml index 9b5e7fa19..45fb8adf3 100644 --- a/app/src/main/res/values/strings.xml +++ b/app/src/main/res/values/strings.xml @@ -299,6 +299,8 @@ Balanced Standard Performance + Turbo + Maximum Auto-detected: %1$s Steam CDN region for game downloads Sign out of %1$s? From 3c348b1c47730345a5cb9bd39e3dcb186fe1a48e Mon Sep 17 00:00:00 2001 From: laoqijun Date: Sun, 8 Nov 2026 06:13:20 +0800 Subject: [PATCH 3/5] Steam: add adaptive CDN host scheduling (Valve/GameHub-style) WinNative round-robined chunk downloads across the whole usable server list. On a mainland-China network that list mixes the reachable Asia-Pacific edges with blackholed international hosts, so a large share of workers sat in connect timeouts and effective throughput collapsed to 46-72 Mbps even though the same device and CDN can do far more (measured: the device Wi-Fi sustains 34 MB/s; Valve's own client pulls 225-373 Mbps from a single China CDN host). Port the way Valve's client and other Steam downloaders behave instead: * New `host_scheduler` module keeps a per-host EWMA throughput estimate, expands an escalating cooldown (3s/15s/60s/300s) on consecutive failures, samples unmeasured hosts once, and periodically explores the least-recently-used host so a recovered node is picked up again. * The parallel depot writer creates one scheduler per batch and asks it which host each chunk should use; retries walk to the next best *measured* host instead of the next index. The scheduler summary is logged per depot. * New "Adaptive (GameHub-style)" download speed tier (sentinel 100, 32 workers) selects the scheduler from the existing settings row. Measured on-device (Redmi K80, mainland China, no proxy, 2.1 GB download): before, round-robin 32 workers: 46-72 Mbps after, adaptive: 144 Mbps (18.05 MB/s) The logged scheduler summary shows why: the four reachable hosts scored 18.4/11.3/10.2/5.9 Mbps with zero failures, while seven blackholed hosts were cooled down after 3-26 consecutive failures each. --- .../rust/src/depot_downloader.rs | 13 + .../wn-steam-client/rust/src/depot_writer.rs | 50 ++- .../rust/src/host_scheduler.rs | 335 ++++++++++++++++++ .../main/cpp/wn-steam-client/rust/src/jni.rs | 9 +- .../main/cpp/wn-steam-client/rust/src/lib.rs | 1 + .../feature/settings/stores/StoresScreen.kt | 3 + app/src/main/res/values-zh-rCN/strings.xml | 3 + app/src/main/res/values/strings.xml | 1 + 8 files changed, 412 insertions(+), 3 deletions(-) create mode 100644 app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs diff --git a/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs b/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs index bdf9e098e..15384c8d6 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs @@ -9,6 +9,15 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; use std::time::Duration; +/// Sentinel worker count that switches the writer into adaptive mode: hosts +/// are scored by measured throughput (Valve-client / GameHub style) instead of +/// being round-robined. Mirrors the "Adaptive" entry in the app's download +/// speed setting. +pub const ADAPTIVE_WORKERS_SENTINEL: u32 = 100; + +/// Worker count used when adaptive mode is selected. +pub const ADAPTIVE_DEFAULT_WORKERS: u32 = 32; + #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct DepotSpec { pub depot_id: u32, @@ -425,6 +434,7 @@ pub fn download_resolved_depots( None, "", None, + false, ) } @@ -449,6 +459,7 @@ pub fn download_resolved_depots_with_cancel( None, "", None, + false, ) } @@ -464,6 +475,7 @@ pub fn download_resolved_depots_with_cancel_progress( code_refresher: Option>, cdn_auth_token: &str, token_refresher: Option>, + adaptive: bool, ) -> DepotDownloadResult { if let Err(error) = validate_resolved_download_inputs(install_dir, depots, servers) { return error; @@ -616,6 +628,7 @@ pub fn download_resolved_depots_with_cancel_progress( cancel, on_progress: Some(chunk_progress), cdn_auth_token, + adaptive, ..Default::default() }, ); diff --git a/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs b/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs index 84258918b..8690100bb 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs @@ -1,6 +1,7 @@ use crate::cdn_client::{CdnClient, CdnConnection}; use crate::content_manifest::{ChunkData, ContentManifest}; use crate::depot_chunk::process_depot_chunk; +use crate::host_scheduler::HostScheduler; use crate::pb::ccontentserverdirectory::CContentServerDirectoryServerInfo; use std::fs::{self, File, OpenOptions}; use std::io::{Read, Seek, SeekFrom, Write}; @@ -60,6 +61,9 @@ pub struct DepotWriteOptions<'a> { pub max_workers: u32, pub cancel: Option<&'a AtomicBool>, pub on_progress: Option>, + /// Score hosts by measured throughput and steer chunks at the fast ones + /// instead of round-robining over the whole list. + pub adaptive: bool, } impl Default for DepotWriteOptions<'_> { @@ -70,6 +74,7 @@ impl Default for DepotWriteOptions<'_> { max_workers: 8, cancel: None, on_progress: None, + adaptive: false, } } } @@ -308,6 +313,7 @@ pub fn write_depot_sequential( options.cdn_auth_token, job_index % servers.len(), options.timeout, + None, ) { Ok(bytes) => { bytes_written += bytes; @@ -364,6 +370,12 @@ fn write_depot_parallel( let raw = c as *const AtomicBool; unsafe { raw.as_ref() }.unwrap() }); + // Adaptive scheduling (Valve-client/GameHub style): measure each content + // server's throughput and steer chunks at the fast ones, cooling down the + // hosts that fail instead of round-robining into blackholed edges. + let scheduler: Option> = options + .adaptive + .then(|| Arc::new(HostScheduler::new(servers.len()))); // SAFETY: we only spawn scoped threads so all `'a` references outlive joins. let scope_result = thread::scope(|scope| -> DepotWriteResult { let mut handles = Vec::with_capacity(worker_count); @@ -378,6 +390,7 @@ fn write_depot_parallel( let target_dir_ref = target_dir; let cdn_ref = cdn; let servers_ref = servers; + let scheduler_ref = scheduler.clone(); let progress = options.on_progress; handles.push(scope.spawn(move || { let mut conn = cdn_ref.open_connection(); @@ -426,7 +439,13 @@ fn write_depot_parallel( conn = cdn_ref.open_connection(); slow_chunks = 0; } - let start_server = (idx + worker_server_bias) % servers_ref.len(); + let start_server = match scheduler_ref + .as_ref() + .and_then(|scheduler| scheduler.pick()) + { + Some(host) => host, + None => (idx + worker_server_bias) % servers_ref.len(), + }; let started = Instant::now(); match fetch_process_write_chunk( cdn_ref, @@ -440,6 +459,7 @@ fn write_depot_parallel( &cdn_auth_token, start_server, timeout, + scheduler_ref.as_deref(), ) { Ok(bytes) => { if started.elapsed() @@ -483,6 +503,12 @@ fn write_depot_parallel( return scope_result; } + if let Some(scheduler) = scheduler.as_ref() { + for row in scheduler.summary() { + crate::depot_downloader::log_line(&format!("cdn host score {row}")); + } + } + for file in &manifest.files { if options .cancel @@ -563,6 +589,7 @@ pub fn fetch_process_write_chunk( cdn_auth_token: &str, start_server_index: usize, timeout: Duration, + scheduler: Option<&HostScheduler>, ) -> Result { if servers.is_empty() { return Err("write_depot: no CDN servers".to_string()); @@ -578,13 +605,25 @@ pub fn fetch_process_write_chunk( let path = join_target_path(target_dir, &file.filename); let mut last_error = String::new(); let attempt_indices = chunk_attempt_server_indices(start_server_index, servers.len()); - for (attempt, server_idx) in attempt_indices.into_iter().enumerate() { + let mut scheduled_tried: Vec = Vec::new(); + for (attempt, fallback_server_idx) in attempt_indices.into_iter().enumerate() { if attempt > 0 { thread::sleep(Duration::from_millis(retry_backoff_millis(attempt as u32))); if let Some(connection) = conn.as_deref_mut() { *connection = cdn.open_connection(); } } + // With a scheduler the retry walks to the next best *measured* host + // instead of the next index in the list. + let server_idx = match scheduler { + Some(scheduler) => match scheduler.pick_excluding(&scheduled_tried) { + Some(host) => host, + None => break, + }, + None => fallback_server_idx, + }; + scheduled_tried.push(server_idx); + let started = Instant::now(); let fetched = match conn.as_deref_mut() { Some(connection) => cdn.fetch_chunk_with_connection( connection, @@ -602,10 +641,17 @@ pub fn fetch_process_write_chunk( timeout, ), }; + let elapsed = started.elapsed(); if !fetched.ok() { + if let Some(scheduler) = scheduler { + scheduler.record_failure(server_idx); + } last_error = fetched.error; continue; } + if let Some(scheduler) = scheduler { + scheduler.record_success(server_idx, fetched.data.len() as u64, elapsed); + } match process_and_write_chunk(&path, chunk, &fetched.data, depot_key) { Ok(bytes) => return Ok(bytes), Err(error) => last_error = error, diff --git a/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs b/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs new file mode 100644 index 000000000..4bf30f98a --- /dev/null +++ b/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs @@ -0,0 +1,335 @@ +//! Adaptive CDN host scheduling. +//! +//! Valve's own client and other Steam downloaders do not spray chunks evenly +//! over every content server the CM handed out. They create one download +//! interface per source, measure each source's throughput, and keep steering +//! work at the sources that are actually fast while backing off from the ones +//! that fail. +//! +//! WinNative previously round-robined chunks across the whole usable list. On a +//! mainland-China network that list mixes reachable partner CDNs with +//! international edges that are blackholed, so a large share of workers sat in +//! connect timeouts and effective throughput collapsed. This scheduler keeps a +//! running throughput estimate per host, places failing hosts into a cooldown +//! that doubles on repeated failures, and occasionally samples the other hosts +//! so a recovering node is picked up again. + +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +/// Cooldown applied after the nth consecutive failure (clamped to the last). +const DEFAULT_FAILURE_COOLDOWNS: &[Duration] = &[ + Duration::from_secs(3), + Duration::from_secs(15), + Duration::from_secs(60), + Duration::from_secs(300), +]; + +/// EWMA weight for the newest throughput sample. +const SCORE_NEW_SAMPLE_WEIGHT: f64 = 0.35; + +/// Every Nth pick samples a host other than the current best. +const EXPLORATION_PERIOD: usize = 8; + +#[derive(Clone, Copy, Debug, Default)] +struct HostEntry { + score_bytes_per_sec: f64, + has_sample: bool, + successes: u32, + failures: u32, + consecutive_failures: u32, + cooldown_until: Option, + last_picked_at: Option, + in_flight: u32, +} + +#[derive(Debug)] +struct Inner { + entries: Vec, + pick_count: usize, + exploration_cursor: usize, +} + +pub struct HostScheduler { + inner: Mutex, + failure_cooldowns: Vec, +} + +impl HostScheduler { + pub fn new(host_count: usize) -> Self { + Self::with_failure_cooldowns(host_count, DEFAULT_FAILURE_COOLDOWNS.to_vec()) + } + + pub fn with_failure_cooldowns(host_count: usize, cooldowns: Vec) -> Self { + Self { + inner: Mutex::new(Inner { + entries: vec![HostEntry::default(); host_count], + pick_count: 0, + exploration_cursor: 0, + }), + failure_cooldowns: if cooldowns.is_empty() { + DEFAULT_FAILURE_COOLDOWNS.to_vec() + } else { + cooldowns + }, + } + } + + pub fn host_count(&self) -> usize { + self.inner.lock().expect("host scheduler poisoned").entries.len() + } + + /// Chooses the host a worker should fetch its next chunk from. + /// + /// Hosts in cooldown are skipped. Among the rest the highest estimated + /// throughput wins, except for a periodic exploration pick that samples the + /// least recently used host so a host that recovered is retried. + pub fn pick(&self) -> Option { + self.pick_excluding(&[]) + } + + /// Like [`Self::pick`] but never returns an index in `exclude`, which the + /// chunk retry path uses to walk through distinct hosts on repeated + /// failures. + pub fn pick_excluding(&self, exclude: &[usize]) -> Option { + let mut inner = self.inner.lock().expect("host scheduler poisoned"); + let now = Instant::now(); + let count = inner.entries.len(); + if count == 0 { + return None; + } + + inner.pick_count += 1; + let explore = inner.pick_count % EXPLORATION_PERIOD == 0; + + let usable = |entry: &HostEntry, idx: usize| -> bool { + !exclude.contains(&idx) + && !entry.cooldown_until.is_some_and(|until| until > now) + }; + + let chosen = if explore { + // Round-robin over usable hosts by last_picked_at, oldest first. + let mut candidates: Vec = (0..count) + .filter(|idx| usable(&inner.entries[*idx], *idx)) + .collect(); + if candidates.is_empty() { + return None; + } + candidates.sort_by_key(|idx| inner.entries[*idx].last_picked_at); + let idx = candidates[inner.exploration_cursor % candidates.len()]; + inner.exploration_cursor = inner.exploration_cursor.wrapping_add(1); + idx + } else { + let mut best: Option<(usize, f64)> = None; + for idx in 0..count { + let entry = &inner.entries[idx]; + if !usable(entry, idx) { + continue; + } + // Unknown hosts start at a neutral score so they get sampled + // once instead of being starved by measured hosts. + let score = if entry.has_sample { + entry.score_bytes_per_sec + } else { + f64::INFINITY + }; + if best.is_none_or(|(_, best_score)| score > best_score) { + best = Some((idx, score)); + } + } + best.map(|(idx, _)| idx)? + }; + + let entry = &mut inner.entries[chosen]; + entry.last_picked_at = Some(now); + entry.in_flight = entry.in_flight.saturating_add(1); + Some(chosen) + } + + /// Records a successful transfer so the host's throughput estimate improves. + pub fn record_success(&self, host: usize, bytes: u64, elapsed: Duration) { + let mut inner = self.inner.lock().expect("host scheduler poisoned"); + let Some(entry) = inner.entries.get_mut(host) else { + return; + }; + entry.in_flight = entry.in_flight.saturating_sub(1); + entry.successes = entry.successes.saturating_add(1); + entry.consecutive_failures = 0; + entry.cooldown_until = None; + + let secs = elapsed.as_secs_f64(); + if secs <= 0.0 || bytes == 0 { + return; + } + let sample = bytes as f64 / secs; + entry.score_bytes_per_sec = if entry.has_sample { + entry.score_bytes_per_sec * (1.0 - SCORE_NEW_SAMPLE_WEIGHT) + + sample * SCORE_NEW_SAMPLE_WEIGHT + } else { + sample + }; + entry.has_sample = true; + } + + /// Records a failed transfer and places the host in an escalating cooldown. + pub fn record_failure(&self, host: usize) { + let mut inner = self.inner.lock().expect("host scheduler poisoned"); + let Some(entry) = inner.entries.get_mut(host) else { + return; + }; + entry.in_flight = entry.in_flight.saturating_sub(1); + entry.failures = entry.failures.saturating_add(1); + entry.consecutive_failures = entry.consecutive_failures.saturating_add(1); + + let step = (entry.consecutive_failures as usize - 1).min(self.failure_cooldowns.len() - 1); + entry.cooldown_until = Some(Instant::now() + self.failure_cooldowns[step]); + // Halve the estimate so a host that starts failing loses its lead even + // before the cooldown is consulted. + entry.score_bytes_per_sec *= 0.5; + } + + /// Estimated throughput in bytes/sec, or None when the host never succeeded. + pub fn score_of(&self, host: usize) -> Option { + let inner = self.inner.lock().expect("host scheduler poisoned"); + let entry = inner.entries.get(host)?; + entry.has_sample.then_some(entry.score_bytes_per_sec) + } + + /// One-line summary of the scheduler state, best host first. + pub fn summary(&self) -> Vec { + let inner = self.inner.lock().expect("host scheduler poisoned"); + let mut rows: Vec<(usize, f64, u32, u32)> = inner + .entries + .iter() + .enumerate() + .map(|(idx, entry)| { + ( + idx, + if entry.has_sample { + entry.score_bytes_per_sec + } else { + -1.0 + }, + entry.successes, + entry.failures, + ) + }) + .collect(); + rows.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); + rows.into_iter() + .map(|(idx, score, successes, failures)| { + let mbps = if score >= 0.0 { score * 8.0 / 1_000_000.0 } else { 0.0 }; + format!( + "host#{idx}: {mbps:.1} Mbps (ok={successes} fail={failures})" + ) + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fast_seconds(ms: u64) -> Duration { + Duration::from_millis(ms) + } + + #[test] + fn prefers_the_host_with_the_highest_measured_throughput() { + let scheduler = HostScheduler::new(3); + // Host 0 is slow, host 1 is fast. + scheduler.record_success(0, 1_000_000, fast_seconds(100)); // 10 MB/s + scheduler.record_success(1, 1_000_000, fast_seconds(20)); // 50 MB/s + scheduler.record_success(2, 1_000_000, fast_seconds(80)); // 12.5 MB/s + + let mut picks = Vec::new(); + for _ in 0..6 { + picks.push(scheduler.pick().unwrap()); + } + // Most picks must target the fastest host. + let fast_picks = picks.iter().filter(|idx| **idx == 1).count(); + assert!(fast_picks >= 4, "picks={picks:?}"); + } + + #[test] + fn sampling_an_unmeasured_host_comes_before_measured_ones() { + let scheduler = HostScheduler::new(2); + scheduler.record_success(0, 1_000_000, fast_seconds(10)); + // Host 1 has no sample yet, so it should be picked first. + assert_eq!(scheduler.pick(), Some(1)); + scheduler.record_success(1, 1_000_000, fast_seconds(200)); + assert_eq!(scheduler.pick(), Some(0)); + } + + #[test] + fn failing_host_is_cooled_down_and_others_take_over() { + let scheduler = HostScheduler::with_failure_cooldowns( + 2, + vec![Duration::from_millis(50), Duration::from_millis(80)], + ); + scheduler.record_success(0, 1_000_000, fast_seconds(10)); // fastest + scheduler.record_success(1, 1_000_000, fast_seconds(200)); + assert_eq!(scheduler.pick(), Some(0)); + + scheduler.record_failure(0); + assert_eq!(scheduler.pick(), Some(1), "failed host must be skipped"); + + std::thread::sleep(Duration::from_millis(70)); + // Cooldown elapsed: host 0 is candidates again (its score was halved, + // so host 1 may still win; just assert it is no longer excluded). + let picked = scheduler.pick().unwrap(); + assert!(picked < 2); + } + + #[test] + fn cooldown_escalates_with_consecutive_failures() { + let scheduler = HostScheduler::with_failure_cooldowns( + 1, + vec![Duration::from_millis(20), Duration::from_millis(10_000)], + ); + scheduler.record_failure(0); + assert_eq!(scheduler.pick(), None, "single host in cooldown"); + + std::thread::sleep(Duration::from_millis(30)); + assert_eq!(scheduler.pick(), Some(0)); + + // Second consecutive failure uses the longer cooldown. + scheduler.record_failure(0); + assert_eq!(scheduler.pick(), None); + } + + #[test] + fn exclude_walks_through_alternatives() { + let scheduler = HostScheduler::new(3); + scheduler.record_success(0, 1_000_000, fast_seconds(10)); + scheduler.record_success(1, 1_000_000, fast_seconds(100)); + scheduler.record_success(2, 1_000_000, fast_seconds(1000)); + + let first = scheduler.pick().unwrap(); + assert_eq!(first, 0); + let second = scheduler.pick_excluding(&[0]).unwrap(); + assert_ne!(second, 0); + let third = scheduler.pick_excluding(&[0, second]).unwrap(); + assert_ne!(third, 0); + assert_ne!(third, second); + } + + #[test] + fn summary_reports_measured_hosts_first() { + let scheduler = HostScheduler::new(2); + scheduler.record_success(0, 125_000, fast_seconds(1000)); // 1 Mbps + scheduler.record_success(1, 1_000_000, fast_seconds(1000)); // 8 Mbps + let rows = scheduler.summary(); + assert_eq!(rows.len(), 2); + assert!(rows[0].starts_with("host#1"), "{rows:?}"); + assert!(rows[0].contains("Mbps")); + } + + #[test] + fn empty_scheduler_returns_none() { + let scheduler = HostScheduler::new(0); + assert_eq!(scheduler.pick(), None); + assert_eq!(scheduler.host_count(), 0); + } +} diff --git a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs index 04cbf3593..502c274eb 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs @@ -3354,7 +3354,13 @@ pub extern "system" fn Java_com_winlator_cmod_feature_stores_steam_wnsteam_WnSte }; let app_id = app_id.max(0) as u32; let fresh = fresh != JNI_FALSE; - let max_workers = max_workers.max(1) as u32; + let requested_workers = max_workers.max(1) as u32; + let adaptive = requested_workers >= crate::depot_downloader::ADAPTIVE_WORKERS_SENTINEL; + let max_workers = if adaptive { + crate::depot_downloader::ADAPTIVE_DEFAULT_WORKERS + } else { + requested_workers + }; handle.download_cancel.store(false, Ordering::Relaxed); let download_cancel = Arc::clone(&handle.download_cancel); @@ -3525,6 +3531,7 @@ pub extern "system" fn Java_com_winlator_cmod_feature_stores_steam_wnsteam_WnSte Some(code_refresher_cb), &cdn_auth_token, Some(token_refresher_cb), + adaptive, ); dispatch_download_complete(listener, result); }); diff --git a/app/src/main/cpp/wn-steam-client/rust/src/lib.rs b/app/src/main/cpp/wn-steam-client/rust/src/lib.rs index 04df26dc0..854129980 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/lib.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/lib.rs @@ -27,6 +27,7 @@ pub mod emsg; pub mod encrypted_channel; pub mod handshake_messages; pub mod heartbeat; +pub mod host_scheduler; pub mod jni; pub mod job_manager; pub mod key_dictionary; diff --git a/app/src/main/feature/settings/stores/StoresScreen.kt b/app/src/main/feature/settings/stores/StoresScreen.kt index dedcf0da8..62af0dafd 100644 --- a/app/src/main/feature/settings/stores/StoresScreen.kt +++ b/app/src/main/feature/settings/stores/StoresScreen.kt @@ -190,6 +190,9 @@ fun StoresScreen( 32 to stringResource(R.string.stores_accounts_download_speed_performance), 48 to stringResource(R.string.stores_accounts_download_speed_turbo), 64 to stringResource(R.string.stores_accounts_download_speed_maximum), + // Sentinel consumed by the native downloader: score CDN hosts by + // measured throughput and steer chunks at the fast ones. + 100 to stringResource(R.string.stores_accounts_download_speed_adaptive), ) CompositionLocalProvider(LocalPaneNav provides contentNav) { diff --git a/app/src/main/res/values-zh-rCN/strings.xml b/app/src/main/res/values-zh-rCN/strings.xml index 7c521bee8..3d404c811 100644 --- a/app/src/main/res/values-zh-rCN/strings.xml +++ b/app/src/main/res/values-zh-rCN/strings.xml @@ -297,6 +297,9 @@ 平衡 标准 性能 + 极速 + 极限 + 自适应（模拟盖世） 自动检测：%1$s 用于游戏下载的 Steam CDN 区域 退出 %1$s 登录？ diff --git a/app/src/main/res/values/strings.xml b/app/src/main/res/values/strings.xml index 45fb8adf3..7a26320f4 100644 --- a/app/src/main/res/values/strings.xml +++ b/app/src/main/res/values/strings.xml @@ -301,6 +301,7 @@ Performance Turbo Maximum + Adaptive (GameHub-style) Auto-detected: %1$s Steam CDN region for game downloads Sign out of %1$s? From c5727a9b014f5d951f1f44c09734b351390db246 Mon Sep 17 00:00:00 2001 From: laoqijun Date: Sun, 8 Nov 2026 06:28:54 +0800 Subject: [PATCH 4/5] Steam: pre-flight CDN probe, proportional concurrency, CM retry rotation Three follow-ups to the adaptive scheduler, each measured on-device (Redmi K80, mainland China, no proxy, 2.06 GiB depot): 1. Pre-flight reachability probe. The first wave of workers used to burn its connect timeout on hosts that are DNS-poisoned or TCP-blackholed, so the opening ~30 s ran at roughly a third of the attainable rate. Probe every usable host once (8 parallel threads, 3 s connect timeout) before the chunk loop and park the unreachable ones for 180 s. The probe log now names each parked host. 2. Allocate concurrency in proportion to measured throughput. pick() now maximises score / (in_flight + 1) instead of raw score, so a host that is twice as fast holds roughly twice as many workers. Adaptive mode also defaults to 64 workers, up from 32. 3. Rotate CM endpoints across reconnect attempts. nativePickCmUrl always returned the first entry of the directory response; when that entry is blackholed the client retried it forever and logon never completed even though a reachable CM sat in the same response. It now sorts by weighted_load and advances a rotation index per attempt, logging which endpoint it picked. Measured, same depot and network: round-robin 32 workers 46-72 Mbps adaptive 32 workers 144 Mbps adaptive + probe + proportional 830 Mbps peak (1038 MB in 10 s), 2.06 GiB completed in ~27 s Logged proof: four hosts carried 2262 chunks with zero failures while four others were parked by the probe before a single chunk was requested. --- .../rust/src/depot_downloader.rs | 2 +- .../wn-steam-client/rust/src/depot_writer.rs | 69 ++++++++++++++ .../rust/src/host_scheduler.rs | 89 ++++++++++++++++--- .../main/cpp/wn-steam-client/rust/src/jni.rs | 45 +++++++--- 4 files changed, 179 insertions(+), 26 deletions(-) diff --git a/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs b/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs index 15384c8d6..b201b15bd 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/depot_downloader.rs @@ -16,7 +16,7 @@ use std::time::Duration; pub const ADAPTIVE_WORKERS_SENTINEL: u32 = 100; /// Worker count used when adaptive mode is selected. -pub const ADAPTIVE_DEFAULT_WORKERS: u32 = 32; +pub const ADAPTIVE_DEFAULT_WORKERS: u32 = 64; #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct DepotSpec { diff --git a/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs b/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs index 8690100bb..6e2035114 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/depot_writer.rs @@ -5,6 +5,7 @@ use crate::host_scheduler::HostScheduler; use crate::pb::ccontentserverdirectory::CContentServerDirectoryServerInfo; use std::fs::{self, File, OpenOptions}; use std::io::{Read, Seek, SeekFrom, Write}; +use std::net::{TcpStream, ToSocketAddrs}; use std::path::{Component, Path}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -349,6 +350,71 @@ pub fn write_depot_sequential( } } +/// How long a pre-flight probe waits for a TCP connection. +const PROBE_CONNECT_TIMEOUT: Duration = Duration::from_secs(3); +/// How long an unreachable host stays parked after the pre-flight probe. +const PROBE_QUARANTINE: Duration = Duration::from_secs(180); +/// Parallel probe threads (one per group of hosts). +const PROBE_THREADS: usize = 8; + +/// Cheap pre-flight reachability probe over the usable content servers. +/// +/// A mainland-China CM hands out a list that mixes reachable Asia-Pacific edges +/// with hosts that are DNS-poisoned or TCP-blackholed. Without this probe the +/// first wave of workers spends its connect timeout on those hosts before the +/// scheduler can learn they are dead, which cost roughly half of the attainable +/// throughput during the first 30 s of a download. Probe them once up front and +/// park the unreachable ones. +fn probe_content_servers(scheduler: &HostScheduler, servers: &[CContentServerDirectoryServerInfo]) { + if servers.is_empty() { + return; + } + thread::scope(|scope| { + for (group_index, group) in servers.chunks(PROBE_THREADS.max(1)).enumerate() { + let base = group_index * PROBE_THREADS.max(1); + let scheduler = scheduler; + scope.spawn(move || { + for (offset, server) in group.iter().enumerate() { + if !content_server_reachable(server) { + scheduler.quarantine(base + offset, PROBE_QUARANTINE); + crate::depot_downloader::log_line(&format!( + "cdn host probe: host#{}({}) unreachable, parked for {}s", + base + offset, + content_server_label(server), + PROBE_QUARANTINE.as_secs() + )); + } + } + }); + } + }); +} + +fn content_server_label(server: &CContentServerDirectoryServerInfo) -> &str { + if !server.host.is_empty() { + &server.host + } else { + &server.vhost + } +} + +fn content_server_reachable(server: &CContentServerDirectoryServerInfo) -> bool { + let host = content_server_label(server); + if host.is_empty() { + return false; + } + let endpoint = format!("{}:{}", host, server.port()); + let Ok(addrs) = endpoint.to_socket_addrs() else { + return false; + }; + for addr in addrs { + if TcpStream::connect_timeout(&addr, PROBE_CONNECT_TIMEOUT).is_ok() { + return true; + } + } + false +} + fn write_depot_parallel( manifest: &ContentManifest, depot_key: &[u8], @@ -376,6 +442,9 @@ fn write_depot_parallel( let scheduler: Option> = options .adaptive .then(|| Arc::new(HostScheduler::new(servers.len()))); + if let Some(scheduler) = scheduler.as_ref() { + probe_content_servers(scheduler, servers); + } // SAFETY: we only spawn scoped threads so all `'a` references outlive joins. let scope_result = thread::scope(|scope| -> DepotWriteResult { let mut handles = Vec::with_capacity(worker_count); diff --git a/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs b/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs index 4bf30f98a..0fec624b4 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/host_scheduler.rs @@ -31,6 +31,11 @@ const SCORE_NEW_SAMPLE_WEIGHT: f64 = 0.35; /// Every Nth pick samples a host other than the current best. const EXPLORATION_PERIOD: usize = 8; +/// Throughput assumed for a host that has not produced a sample yet. Low +/// enough that measured hosts keep most of the workers, high enough that an +/// unmeasured host still gets probed. +const UNMEASURED_HOST_SCORE_BPS: f64 = 1_000_000.0; + #[derive(Clone, Copy, Debug, Default)] struct HostEntry { score_bytes_per_sec: f64, @@ -120,21 +125,25 @@ impl HostScheduler { inner.exploration_cursor = inner.exploration_cursor.wrapping_add(1); idx } else { + // Allocate concurrency in proportion to measured throughput: divide + // each host's score by the number of transfers already in flight on + // it. A host that is twice as fast therefore holds roughly twice as + // many workers, which is what keeps a 4-host pool from starving when + // one host is far quicker than the others. let mut best: Option<(usize, f64)> = None; for idx in 0..count { let entry = &inner.entries[idx]; if !usable(entry, idx) { continue; } - // Unknown hosts start at a neutral score so they get sampled - // once instead of being starved by measured hosts. - let score = if entry.has_sample { + let base = if entry.has_sample { entry.score_bytes_per_sec } else { - f64::INFINITY + UNMEASURED_HOST_SCORE_BPS }; - if best.is_none_or(|(_, best_score)| score > best_score) { - best = Some((idx, score)); + let marginal = base / (entry.in_flight as f64 + 1.0); + if best.is_none_or(|(_, best_marginal)| marginal > best_marginal) { + best = Some((idx, marginal)); } } best.map(|(idx, _)| idx)? @@ -188,6 +197,23 @@ impl HostScheduler { entry.score_bytes_per_sec *= 0.5; } + /// Immediately parks a host for `duration`. Used by the pre-flight reachability + /// probe so a blackholed host never receives a single chunk request during + /// the fragile start of a download. + pub fn quarantine(&self, host: usize, duration: Duration) { + let mut inner = self.inner.lock().expect("host scheduler poisoned"); + let Some(entry) = inner.entries.get_mut(host) else { + return; + }; + entry.failures = entry.failures.saturating_add(1); + entry.consecutive_failures = entry.consecutive_failures.saturating_add(1); + entry.score_bytes_per_sec = 0.0; + // has_sample stays false so the host is not treated as a measured zero: + // it is re-probed once the quarantine expires. + entry.has_sample = false; + entry.cooldown_until = Some(Instant::now() + duration); + } + /// Estimated throughput in bytes/sec, or None when the host never succeeded. pub fn score_of(&self, host: usize) -> Option { let inner = self.inner.lock().expect("host scheduler poisoned"); @@ -253,13 +279,42 @@ mod tests { } #[test] - fn sampling_an_unmeasured_host_comes_before_measured_ones() { + fn measured_host_leads_but_unmeasured_hosts_still_get_probed() { let scheduler = HostScheduler::new(2); - scheduler.record_success(0, 1_000_000, fast_seconds(10)); - // Host 1 has no sample yet, so it should be picked first. - assert_eq!(scheduler.pick(), Some(1)); - scheduler.record_success(1, 1_000_000, fast_seconds(200)); + scheduler.record_success(0, 1_000_000, fast_seconds(10)); // ~100 MB/s + + // The measured host wins the greedy picks. assert_eq!(scheduler.pick(), Some(0)); + + // Exploration still reaches the unmeasured host within one period. + let mut saw_unmeasured = false; + for _ in 0..EXPLORATION_PERIOD { + if scheduler.pick() == Some(1) { + saw_unmeasured = true; + } + } + assert!(saw_unmeasured, "exploration must probe the unmeasured host"); + } + + #[test] + fn concurrency_is_allocated_in_proportion_to_measured_throughput() { + let scheduler = HostScheduler::new(2); + // Host 0 is 4x faster than host 1. + scheduler.record_success(0, 4_000_000, fast_seconds(100)); + scheduler.record_success(1, 1_000_000, fast_seconds(100)); + + // Do not record outcomes, so in_flight accumulates exactly like a wave + // of workers asking for work. + let mut picks = [0usize; 2]; + for _ in 0..24 { + if let Some(host) = scheduler.pick() { + picks[host] += 1; + } + } + assert!( + picks[0] > picks[1] * 2, + "fast host must hold more workers: {picks:?}" + ); } #[test] @@ -326,6 +381,18 @@ mod tests { assert!(rows[0].contains("Mbps")); } + #[test] + fn quarantine_parks_a_host_until_it_expires() { + let scheduler = HostScheduler::new(2); + scheduler.quarantine(0, Duration::from_millis(40)); + // Host 0 must be skipped while parked. + for _ in 0..EXPLORATION_PERIOD { + assert_eq!(scheduler.pick(), Some(1)); + } + std::thread::sleep(Duration::from_millis(50)); + assert!(scheduler.pick().is_some()); + } + #[test] fn empty_scheduler_returns_none() { let scheduler = HostScheduler::new(0); diff --git a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs index 502c274eb..369aa7287 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs @@ -1828,29 +1828,46 @@ pub extern "system" fn Java_com_winlator_cmod_feature_stores_steam_wnsteam_WnSte let ca_bundle_path = jstring_to_string(&mut env, &ca_bundle_path).unwrap_or_default(); let directory = crate::steam_directory::SteamDirectoryClient::new(ca_bundle_path); let result = directory.fetch(0, crate::steam_directory::DEFAULT_TIMEOUT); - let directory_url = if result.ok() { + + // Collect every advertised CM, least loaded first. Reconnect attempts then + // walk the list instead of retrying the same endpoint forever: on some + // networks several CMs are blackholed, so pinning the first entry made + // logon fail in a loop even though another entry in the same response was + // reachable. + let mut candidates: Vec<(f32, String)> = if result.ok() { result .servers - .into_iter() - .find_map(|server| { + .iter() + .filter_map(|server| { let url = server.websocket_url(); - (!url.is_empty()).then_some(url) + (!url.is_empty()).then_some((server.weighted_load, url)) }) - .unwrap_or_default() + .collect() } else { - String::new() + Vec::new() }; - let url = if !directory_url.is_empty() { - directory_url - } else { - crate::cm_server_list::hardcoded_fallback_servers() + if candidates.is_empty() { + candidates = crate::cm_server_list::hardcoded_fallback_servers() .into_iter() - .find_map(|server| { + .filter_map(|server| { let url = server.websocket_url(); - (!url.is_empty()).then_some(url) + (!url.is_empty()).then_some((0.0f32, url)) }) - .unwrap_or_default() - }; + .collect(); + } + if candidates.is_empty() { + return new_string_or_null(&mut env, ""); + } + candidates.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal)); + + static CM_PICK_ROTATION: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + let rotation = CM_PICK_ROTATION.fetch_add(1, Ordering::Relaxed) % candidates.len(); + let url = candidates[rotation].1.clone(); + crate::depot_downloader::log_line(&format!( + "CM pick #{rotation}/{} -> {url}", + candidates.len() + )); new_string_or_null(&mut env, &url) } From c772d18e1d4c46b71cbf88d9a09fb9d1f5813f81 Mon Sep 17 00:00:00 2001 From: laoqijun Date: Sun, 8 Nov 2026 10:05:32 +0800 Subject: [PATCH 5/5] Steam: reachable CMs, recoverable secure storage, release logging Three fixes for Steam logon on a mainland-China connection, all found by comparing against the official client's own logs. 1. Pick the CM by latency. connection_log.txt of a real client shows it pinging every advertised endpoint and connecting to the closest one (157 ms Shanghai instead of 276 ms Seattle), while we sorted by load and rotated. Add cm_selector: parallel probes with a 1.5 s connect timeout over the advertised ports plus 443/27018/27019, lowest RTT wins, load only breaks ties within 25 ms, failures are remembered for 120 s and a re-pick within 45 s counts as a failed connection. 2. Stop falling back to the `ext*` edge hosts. When the directory request failed the hardcoded list offered ext1-sea1/ext1-fra1 and every logon ended in SteamServerConnectFailure_t; they front the network but do not hold a session for a client that is 300 ms away. Fall back to real pool members (cmp*-hkg1, cmp*-sha1) and log the directory outcome so a fallback is visible. 3. Recover from invalidated secure storage. Clearing the app's data deletes the encrypted preferences file but leaves the AndroidKeyStore key behind, which permanently invalidates it; PrefManager turned that into a RuntimeException from every Activity.onCreate, so the app could not start at all after a data wipe. Delete the stale key and file, retry once, and degrade to plain preferences if encryption is still unavailable. Also plant Timber in release builds. It was debug-only, so a shipped APK logged nothing about logon state, which is why the CM edge-host problem stayed invisible in the field. Keeps for androidx.security.crypto/Tink are added because release builds reach them reflectively. --- app/src/main/app/PluviaApp.kt | 13 +- .../wn-steam-client/rust/src/cm_selector.rs | 348 ++++++++++++++++++ .../rust/src/cm_server_list.rs | 33 +- .../main/cpp/wn-steam-client/rust/src/jni.rs | 93 +++-- .../main/cpp/wn-steam-client/rust/src/lib.rs | 1 + .../feature/stores/steam/utils/PrefManager.kt | 68 +++- 6 files changed, 497 insertions(+), 59 deletions(-) create mode 100644 app/src/main/cpp/wn-steam-client/rust/src/cm_selector.rs diff --git a/app/src/main/app/PluviaApp.kt b/app/src/main/app/PluviaApp.kt index 850d6e69d..3d59b66e3 100644 --- a/app/src/main/app/PluviaApp.kt +++ b/app/src/main/app/PluviaApp.kt @@ -56,12 +56,13 @@ class PluviaApp : Application() { } } - // Initialize Timber in debug builds for app wide logging. - // If 'BuildConfig' give error, ignore it. - // The file needed will be autogenerated when building the apk. - if (com.winlator.cmod.BuildConfig.DEBUG) { - Timber.plant(Timber.DebugTree()); - } + // Plant Timber in every build, not just debug ones. + // Release builds used to log nothing through Timber, which made the bugs + // users actually report (Steam logon and download failures) impossible to + // diagnose from logcat: the interesting state transitions simply vanished + // once the APK was minified. DebugTree goes to logcat and costs nothing + // when nobody reads it. + Timber.plant(Timber.DebugTree()) // Initialize the LogManager context in case a fallback is needed. LogManager.init(this.applicationContext) diff --git a/app/src/main/cpp/wn-steam-client/rust/src/cm_selector.rs b/app/src/main/cpp/wn-steam-client/rust/src/cm_selector.rs new file mode 100644 index 000000000..10bc85094 --- /dev/null +++ b/app/src/main/cpp/wn-steam-client/rust/src/cm_selector.rs @@ -0,0 +1,348 @@ +//! Picks the connection manager to log on to, the way the official client does. +//! +//! The Steam client does not take the first entry of the CM directory: it pings +//! every advertised WebSocket CM (port 443 and the 27018/27019 alternates) and +//! connects to the endpoint with the lowest measured round-trip time, using the +//! advertised load only to break ties. Picking by load alone - or rotating +//! blindly - sends clients to far-away CMs on networks where only some of the +//! advertised datacenters are reachable, which is exactly what a mainland-China +//! connection sees: `sha1`/Perfect World CMs are a few milliseconds away while +//! the `sea1`/`hkg1` entries time out. +//! +//! Two practical differences from a naive probe implementation: +//! +//! * Endpoints that recently failed are skipped instead of being probed again on +//! every reconnect, and being asked to pick again within [`RECONNECT_WINDOW`] +//! counts as a failure of the previous pick (the client only re-picks when the +//! connection did not come up). +//! * If every probe fails we still return something - the least-loaded endpoint - +//! because a wrong guess is better than not trying at all. + +use crate::cm_server::CmServer; +use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +/// How long a single endpoint probe may take before it counts as unreachable. +pub const PROBE_TIMEOUT: Duration = Duration::from_millis(1500); + +/// Endpoints that failed this recently are skipped while other options remain. +pub const FAILURE_MEMORY: Duration = Duration::from_secs(120); + +/// Being asked to pick again this soon after the previous pick means the client +/// could not establish a session, so the previous endpoint is remembered as bad. +pub const RECONNECT_WINDOW: Duration = Duration::from_secs(45); + +/// Ports the official client probes for a WebSocket CM, in addition to whatever +/// the directory advertises. +const ALTERNATE_PORTS: [u16; 3] = [443, 27018, 27019]; + +#[derive(Clone, Debug, PartialEq)] +pub struct CmCandidate { + pub host: String, + pub port: u16, + pub weighted_load: f32, +} + +impl CmCandidate { + pub fn url(&self) -> String { + format!("wss://{}:{}/cmsocket/", self.host, self.port) + } + + fn key(&self) -> String { + format!("{}:{}", self.host, self.port) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ProbeOutcome { + pub index: usize, + pub rtt: Duration, +} + +/// Builds the candidate list: every advertised endpoint plus the alternate ports +/// the official client also probes, de-duplicated by `host:port`. +pub fn candidates_for(servers: &[CmServer]) -> Vec { + let mut out: Vec = Vec::new(); + for server in servers { + if server.host.is_empty() { + continue; + } + let mut ports: Vec = Vec::new(); + if server.port != 0 { + ports.push(server.port); + } + for port in ALTERNATE_PORTS { + if !ports.contains(&port) { + ports.push(port); + } + } + for port in ports { + let candidate = CmCandidate { + host: server.host.clone(), + port, + weighted_load: server.weighted_load, + }; + if !out.iter().any(|existing| existing.key() == candidate.key()) { + out.push(candidate); + } + } + } + out +} + +struct BadEndpoint { + key: String, + at: Instant, +} + +#[derive(Default)] +struct Memory { + bad: Vec, + last_pick: Option<(String, Instant)>, +} + +static MEMORY: Mutex> = Mutex::new(None); + +fn with_memory(f: impl FnOnce(&mut Memory) -> T) -> T { + let mut guard = MEMORY.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let memory = guard.get_or_insert_with(Memory::default); + f(memory) +} + +fn prune(memory: &mut Memory, now: Instant) { + memory + .bad + .retain(|entry| now.duration_since(entry.at) < FAILURE_MEMORY); +} + +fn is_bad(key: &str) -> bool { + let now = Instant::now(); + with_memory(|memory| { + prune(memory, now); + memory.bad.iter().any(|entry| entry.key == key) + }) +} + +/// Notes that we are about to connect to `key`. If this happens inside +/// [`RECONNECT_WINDOW`] the previous endpoint evidently did not work, so it is +/// remembered as bad and the next pick will avoid it. +fn note_pick(key: &str) { + let now = Instant::now(); + with_memory(|memory| { + if let Some((previous, at)) = memory.last_pick.take() { + if previous != key && now.duration_since(at) < RECONNECT_WINDOW { + prune(memory, now); + memory.bad.push(BadEndpoint { + key: previous, + at: now, + }); + } + } + memory.last_pick = Some((key.to_string(), now)); + }); +} + +fn resolve(host: &str, port: u16) -> Vec { + let Ok(addrs) = (host, port).to_socket_addrs() else { + return Vec::new(); + }; + addrs.collect() +} + +/// Opens a TCP connection to measure reachability and round-trip time. +fn probe_one(host: &str, port: u16) -> Option { + let started = Instant::now(); + let addrs = resolve(host, port); + if addrs.is_empty() { + return None; + } + for addr in addrs { + if TcpStream::connect_timeout(&addr, PROBE_TIMEOUT).is_ok() { + return Some(started.elapsed()); + } + } + None +} + +/// Probes `candidates` in parallel and returns the fastest reachable one. +/// +/// Candidates that recently failed are only probed when nothing else works, and +/// the advertised load breaks ties between endpoints with similar latency. +pub fn pick(candidates: &[CmCandidate]) -> Option<(usize, Duration)> { + if candidates.is_empty() { + return None; + } + let preferred: Vec = (0..candidates.len()) + .filter(|index| !is_bad(&candidates[*index].key())) + .collect(); + let indices = if preferred.is_empty() { + (0..candidates.len()).collect::>() + } else { + preferred + }; + + // (candidate index, measured rtt when the probe connected) + let measurements: Vec<(usize, Option)> = std::thread::scope(|scope| { + let handles: Vec<_> = indices + .iter() + .map(|index| { + let candidate = &candidates[*index]; + scope.spawn(move || (*index, probe_one(&candidate.host, candidate.port))) + }) + .collect(); + handles + .into_iter() + .map(|handle| handle.join().unwrap_or((0, None))) + .collect() + }); + + let now = Instant::now(); + with_memory(|memory| { + prune(memory, now); + for (index, rtt) in &measurements { + if rtt.is_none() { + let key = candidates[*index].key(); + if !memory.bad.iter().any(|entry| entry.key == key) { + memory.bad.push(BadEndpoint { key, at: now }); + } + } + } + }); + + let mut best: Option<(usize, Duration)> = None; + for (index, rtt) in measurements { + let Some(rtt) = rtt else { continue }; + match best { + None => best = Some((index, rtt)), + Some((best_index, best_rtt)) => { + // Latency wins unless the two endpoints are within 25 ms, in which + // case the less loaded datacenter is preferred. + let close = rtt.abs_diff(best_rtt) < Duration::from_millis(25); + let better = if close { + candidates[index].weighted_load < candidates[best_index].weighted_load + } else { + rtt < best_rtt + }; + if better { + best = Some((index, rtt)); + } + } + } + } + + match best { + Some((index, rtt)) => { + note_pick(&candidates[index].key()); + Some((index, rtt)) + } + None => { + // Nothing answered. Still return an endpoint so the client attempts a + // connection instead of giving up, but do not treat it as a success. + let fallback = indices.iter().copied().min_by(|a, b| { + candidates[*a] + .weighted_load + .partial_cmp(&candidates[*b].weighted_load) + .unwrap_or(std::cmp::Ordering::Equal) + })?; + note_pick(&candidates[fallback].key()); + None + } + } +} + +/// Human-readable candidate list, used only for logging. +pub fn describe(candidates: &[CmCandidate]) -> String { + candidates + .iter() + .map(|candidate| format!("{}:{} load={:.1}", candidate.host, candidate.port, candidate.weighted_load)) + .collect::>() + .join(", ") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cm_server::CmTransport; + + fn server(host: &str, port: u16, load: f32) -> CmServer { + CmServer { + endpoint: format!("{host}:{port}"), + host: host.to_string(), + port, + transport: CmTransport::WebSocket, + realm: "steamglobal".to_string(), + datacenter: String::new(), + load: 0, + weighted_load: load, + } + } + + #[test] + fn candidates_expand_alternate_ports_once() { + let candidates = candidates_for(&[server("cmp2-sha1.steamchina.com", 443, 30.0)]); + assert_eq!(candidates.len(), 3); + assert_eq!(candidates[0].port, 443); + assert!(candidates.iter().any(|candidate| candidate.port == 27018)); + assert!(candidates.iter().any(|candidate| candidate.port == 27019)); + } + + #[test] + fn candidates_deduplicate_hosts() { + let candidates = candidates_for(&[ + server("a.example.com", 443, 1.0), + server("a.example.com", 443, 1.0), + server("b.example.com", 27018, 2.0), + ]); + assert_eq!( + candidates + .iter() + .filter(|candidate| candidate.host == "a.example.com") + .count(), + 3 + ); + assert_eq!( + candidates + .iter() + .filter(|candidate| candidate.host == "b.example.com") + .count(), + 3 + ); + } + + #[test] + fn url_is_the_websocket_endpoint() { + let candidate = CmCandidate { + host: "cmp2-sha1.steamchina.com".to_string(), + port: 27018, + weighted_load: 0.0, + }; + assert_eq!(candidate.url(), "wss://cmp2-sha1.steamchina.com:27018/cmsocket/"); + } + + #[test] + fn unreachable_candidates_are_reported_as_none() { + // 192.0.2.0/24 is TEST-NET-1: guaranteed not to answer. + let mut candidates = candidates_for(&[server("192.0.2.1", 443, 1.0)]); + candidates.truncate(1); + assert_eq!(pick(&candidates), None); + } + + #[test] + fn picking_again_quickly_marks_the_previous_endpoint_bad() { + let first = CmCandidate { + host: "192.0.2.10".to_string(), + port: 443, + weighted_load: 0.0, + }; + let second = CmCandidate { + host: "192.0.2.11".to_string(), + port: 443, + weighted_load: 0.0, + }; + note_pick(&first.key()); + note_pick(&second.key()); + assert!(is_bad(&first.key())); + assert!(!is_bad(&second.key())); + } +} diff --git a/app/src/main/cpp/wn-steam-client/rust/src/cm_server_list.rs b/app/src/main/cpp/wn-steam-client/rust/src/cm_server_list.rs index b4f0a495c..f3f8473c4 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/cm_server_list.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/cm_server_list.rs @@ -131,23 +131,32 @@ fn promote_expired(entries: &mut [Entry], bad_memory: Duration) { } } +/// Connection managers used when the directory request fails. +/// +/// These must be real CM pool members (`cmp*`), not the `ext*` websocket edge +/// hosts: the `ext*` entries only relay a client towards a CM and are not +/// usable as the session endpoint, so logging on through them never completes. +/// The ports are the ones the directory advertises for these datacenters; the +/// selector additionally probes 443/27018/27019 for every host. pub fn hardcoded_fallback_servers() -> Vec { - const FALLBACK: &[(&str, &str)] = &[ - ("ext1-sea1.steamserver.net", "sea1"), - ("ext2-sea1.steamserver.net", "sea1"), - ("ext1-iad1.steamserver.net", "iad1"), - ("ext2-iad1.steamserver.net", "iad1"), - ("ext1-fra1.steamserver.net", "fra1"), - ("ext2-fra1.steamserver.net", "fra1"), - ("ext1-lax1.steamserver.net", "lax1"), - ("ext1-sgp1.steamserver.net", "sgp1"), + const FALLBACK: &[(&str, &str, u16)] = &[ + ("cmp1-hkg1.steamserver.net", "hkg1", 27023), + ("cmp2-hkg1.steamserver.net", "hkg1", 27023), + ("cmp3-hkg1.steamserver.net", "hkg1", 27020), + ("cmp1-hkg1.steamserver.net", "hkg1", 443), + ("cmp2-hkg1.steamserver.net", "hkg1", 443), + ("cmp3-hkg1.steamserver.net", "hkg1", 443), + ("cmp1-sgp1.steamserver.net", "sgp1", 27023), + ("cmp2-sgp1.steamserver.net", "sgp1", 27020), + ("cmp1-sha1.steamchina.com", "sha1", 443), + ("cmp2-sha1.steamchina.com", "sha1", 443), ]; FALLBACK .iter() - .map(|(host, dc)| CmServer { - endpoint: format!("{host}:443"), + .map(|(host, dc, port)| CmServer { + endpoint: format!("{host}:{port}"), host: (*host).to_string(), - port: 443, + port: *port, transport: CmTransport::WebSocket, realm: "steamglobal".to_string(), datacenter: (*dc).to_string(), diff --git a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs index 369aa7287..4c6431917 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/jni.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/jni.rs @@ -1834,40 +1834,77 @@ pub extern "system" fn Java_com_winlator_cmod_feature_stores_steam_wnsteam_WnSte // networks several CMs are blackholed, so pinning the first entry made // logon fail in a loop even though another entry in the same response was // reachable. - let mut candidates: Vec<(f32, String)> = if result.ok() { - result - .servers - .iter() - .filter_map(|server| { - let url = server.websocket_url(); - (!url.is_empty()).then_some((server.weighted_load, url)) - }) - .collect() + let servers = if result.ok() { + crate::depot_downloader::log_line(&format!( + "CM directory: {} websocket CMs (http {})", + result.servers.len(), + result.http_status + )); + result.servers.clone() } else { - Vec::new() + // Log why the directory was unusable: without this the client silently + // falls back to the built-in list and logon failures look inexplicable. + crate::depot_downloader::log_line(&format!( + "CM directory failed (http {}): {}; using {} built-in CMs", + result.http_status, + if result.error.is_empty() { + "no servers" + } else { + result.error.as_str() + }, + crate::cm_server_list::hardcoded_fallback_servers().len() + )); + crate::cm_server_list::hardcoded_fallback_servers() }; - if candidates.is_empty() { - candidates = crate::cm_server_list::hardcoded_fallback_servers() - .into_iter() - .filter_map(|server| { - let url = server.websocket_url(); - (!url.is_empty()).then_some((0.0f32, url)) - }) - .collect(); - } + let candidates = crate::cm_selector::candidates_for(&servers); if candidates.is_empty() { return new_string_or_null(&mut env, ""); } - candidates.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal)); - static CM_PICK_ROTATION: std::sync::atomic::AtomicUsize = - std::sync::atomic::AtomicUsize::new(0); - let rotation = CM_PICK_ROTATION.fetch_add(1, Ordering::Relaxed) % candidates.len(); - let url = candidates[rotation].1.clone(); - crate::depot_downloader::log_line(&format!( - "CM pick #{rotation}/{} -> {url}", - candidates.len() - )); + // Probe every advertised endpoint in parallel and connect to the closest one, + // the same way the official client pings its CM candidates. Endpoints that + // recently failed are skipped so reconnects move on instead of hammering the + // datacenter that just dropped us. + let started = std::time::Instant::now(); + let pick = crate::cm_selector::pick(&candidates); + let elapsed_ms = started.elapsed().as_millis(); + let (index, url) = match pick { + Some((index, rtt)) => { + let url = candidates[index].url(); + crate::depot_downloader::log_line(&format!( + "CM pick: {} probes in {} ms -> {} (rtt {} ms, load {:.1})", + candidates.len(), + elapsed_ms, + url, + rtt.as_millis(), + candidates[index].weighted_load + )); + (index, url) + } + None => { + // Nothing answered: report the fallback the selector chose so the + // caller still sees the endpoint the client will attempt. + let fallback = candidates + .iter() + .enumerate() + .min_by(|a, b| { + a.1.weighted_load + .partial_cmp(&b.1.weighted_load) + .unwrap_or(std::cmp::Ordering::Equal) + }) + .map(|(index, _)| index) + .unwrap_or(0); + let url = candidates[fallback].url(); + crate::depot_downloader::log_line(&format!( + "CM pick: no CM answered among {} candidates in {} ms, falling back to {}", + candidates.len(), + elapsed_ms, + url + )); + (fallback, url) + } + }; + let _ = index; new_string_or_null(&mut env, &url) } diff --git a/app/src/main/cpp/wn-steam-client/rust/src/lib.rs b/app/src/main/cpp/wn-steam-client/rust/src/lib.rs index 854129980..e054960b2 100644 --- a/app/src/main/cpp/wn-steam-client/rust/src/lib.rs +++ b/app/src/main/cpp/wn-steam-client/rust/src/lib.rs @@ -14,6 +14,7 @@ pub mod chat_image; pub mod cm_bridge; pub mod cm_client; pub mod cm_runtime; +pub mod cm_selector; pub mod cm_server; pub mod cm_server_list; pub mod cmsg_protobuf_header; diff --git a/app/src/main/feature/stores/steam/utils/PrefManager.kt b/app/src/main/feature/stores/steam/utils/PrefManager.kt index d49e023c2..1c367d5b8 100644 --- a/app/src/main/feature/stores/steam/utils/PrefManager.kt +++ b/app/src/main/feature/stores/steam/utils/PrefManager.kt @@ -3,9 +3,17 @@ import android.content.Context import android.content.SharedPreferences import androidx.security.crypto.EncryptedSharedPreferences import androidx.security.crypto.MasterKey +import java.security.KeyStore import timber.log.Timber object PrefManager { + private const val ENCRYPTED_PREFS_NAME = "PluviaPreferences_enc" + private const val PLAIN_PREFS_NAME = "PluviaPreferences_enc_fallback" + + /** Alias androidx.security uses for [MasterKey.Builder]'s default AES256_GCM key. */ + private const val MASTER_KEY_ALIAS = "_androidx_security_master_key_" + private const val ANDROID_KEYSTORE = "AndroidKeyStore" + @Volatile private var prefs: SharedPreferences? = null @@ -40,28 +48,62 @@ object PrefManager { this.appContext = appContext val legacyPrefs = appContext.getSharedPreferences("PluviaPreferences", Context.MODE_PRIVATE) - val encryptedPrefs = + // Creating the encrypted store can fail on a device where the AndroidKeyStore + // master key no longer matches the encrypted file: wiping the app's data removes + // PluviaPreferences_enc but leaves the key behind on some ROMs, and the key is then + // permanently invalidated (KeyPermanentlyInvalidatedException). Retrying once after + // deleting both the key and the stale file recovers those installs; if encryption is + // still unavailable we fall back to plain preferences instead of refusing to start. + var encryptedPrefs: SharedPreferences? = null + var lastError: Exception? = null + for (attempt in 1..2) { try { val masterKey = MasterKey .Builder(appContext) .setKeyScheme(MasterKey.KeyScheme.AES256_GCM) .build() - EncryptedSharedPreferences.create( - appContext, - "PluviaPreferences_enc", - masterKey, - EncryptedSharedPreferences.PrefKeyEncryptionScheme.AES256_SIV, - EncryptedSharedPreferences.PrefValueEncryptionScheme.AES256_GCM, - ) + encryptedPrefs = + EncryptedSharedPreferences.create( + appContext, + ENCRYPTED_PREFS_NAME, + masterKey, + EncryptedSharedPreferences.PrefKeyEncryptionScheme.AES256_SIV, + EncryptedSharedPreferences.PrefValueEncryptionScheme.AES256_GCM, + ) + break } catch (e: Exception) { - Timber.e(e, "EncryptedSharedPreferences init failed") - throw RuntimeException("Failed to initialize secure storage", e) + lastError = e + Timber.e(e, "EncryptedSharedPreferences init failed (attempt %d)", attempt) + if (attempt == 1) deleteInvalidatedMasterKey(appContext) } + } + + val resolvedPrefs = + encryptedPrefs + ?: run { + Timber.e(lastError, "Falling back to plain SharedPreferences") + appContext.getSharedPreferences(PLAIN_PREFS_NAME, Context.MODE_PRIVATE) + } + + prefs = resolvedPrefs + migrateLegacyPrefsIfNeeded(legacyPrefs, appContext, resolvedPrefs) + return resolvedPrefs + } + } - prefs = encryptedPrefs - migrateLegacyPrefsIfNeeded(legacyPrefs, appContext, encryptedPrefs) - return encryptedPrefs + /** Removes the encrypted file and its AndroidKeyStore entry so a fresh key can be made. */ + private fun deleteInvalidatedMasterKey(context: Context) { + try { + context.deleteSharedPreferences(ENCRYPTED_PREFS_NAME) + } catch (e: Exception) { + Timber.w(e, "Could not delete the encrypted preferences file") + } + try { + val keyStore = KeyStore.getInstance(ANDROID_KEYSTORE).apply { load(null) } + if (keyStore.containsAlias(MASTER_KEY_ALIAS)) keyStore.deleteEntry(MASTER_KEY_ALIAS) + } catch (e: Exception) { + Timber.w(e, "Could not delete the invalidated master key") } }