Add progress counter for car time downloads

Show a persistent single-line progress counter ("Downloading track times
for car 1234/7589") while fetch_times downloads timing data. A ProgressBar
from indicatif 0.17 is sized to the car list, cloned into each spawned
download task, and incremented after fetch_car completes so it tracks
completions rather than starts; Arc internals make this safe under the
concurrent semaphore. The bar is cleared via finish_and_clear() before
the error/warning logging loop so retry messages don't garble the line,
and indicatif renders nothing when stderr is not a TTY.

Also apply pending rustfmt reformatting to src/capture/windows.rs
(formatting only, no logic changes).
This commit is contained in:
Jan-Peter von Hunnius 2026-08-22 10:53:17 +02:00
parent f829ad52a4
commit cc73fe62fc
4 changed files with 136 additions and 74 deletions

64
Cargo.lock generated
View file

@ -412,6 +412,19 @@ version = "0.4.32"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc14f565cf027a105f7a44ccf9e5b424348421a1d8952a8fc9d499d313107789" checksum = "cc14f565cf027a105f7a44ccf9e5b424348421a1d8952a8fc9d499d313107789"
[[package]]
name = "console"
version = "0.15.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "054ccb5b10f9f2cbf51eb355ca1d05c2d279ce1804688d0db74b4733a5aeafd8"
dependencies = [
"encode_unicode",
"libc",
"once_cell",
"unicode-width",
"windows-sys 0.59.0",
]
[[package]] [[package]]
name = "const-oid" name = "const-oid"
version = "0.10.2" version = "0.10.2"
@ -614,6 +627,12 @@ version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
[[package]]
name = "encode_unicode"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
[[package]] [[package]]
name = "encoding_rs" name = "encoding_rs"
version = "0.8.35" version = "0.8.35"
@ -1240,6 +1259,19 @@ dependencies = [
"serde_core", "serde_core",
] ]
[[package]]
name = "indicatif"
version = "0.17.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "183b3088984b400f4cfac3620d5e076c84da5364016b4f49473de574b2586235"
dependencies = [
"console",
"number_prefix",
"portable-atomic",
"unicode-width",
"web-time",
]
[[package]] [[package]]
name = "ipnet" name = "ipnet"
version = "2.12.0" version = "2.12.0"
@ -1550,6 +1582,12 @@ dependencies = [
"autocfg", "autocfg",
] ]
[[package]]
name = "number_prefix"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3"
[[package]] [[package]]
name = "oid-registry" name = "oid-registry"
version = "0.8.1" version = "0.8.1"
@ -2509,6 +2547,7 @@ dependencies = [
"http", "http",
"http-body-util", "http-body-util",
"hudsucker", "hudsucker",
"indicatif",
"log", "log",
"md-5", "md-5",
"mockito", "mockito",
@ -2673,6 +2712,12 @@ version = "1.0.24"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
[[package]]
name = "unicode-width"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254"
[[package]] [[package]]
name = "unicode-xid" name = "unicode-xid"
version = "0.2.6" version = "0.2.6"
@ -2879,6 +2924,16 @@ dependencies = [
"wasm-bindgen", "wasm-bindgen",
] ]
[[package]]
name = "web-time"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb"
dependencies = [
"js-sys",
"wasm-bindgen",
]
[[package]] [[package]]
name = "webpki-roots" name = "webpki-roots"
version = "0.26.11" version = "0.26.11"
@ -2941,6 +2996,15 @@ dependencies = [
"windows-targets", "windows-targets",
] ]
[[package]]
name = "windows-sys"
version = "0.59.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
dependencies = [
"windows-targets",
]
[[package]] [[package]]
name = "windows-sys" name = "windows-sys"
version = "0.61.2" version = "0.61.2"

View file

@ -29,6 +29,7 @@ thiserror = "2"
dashmap = "6" dashmap = "6"
log = "0.4" log = "0.4"
env_logger = { version = "0.11", optional = true } env_logger = { version = "0.11", optional = true }
indicatif = "0.17"
[target.'cfg(windows)'.dependencies] [target.'cfg(windows)'.dependencies]
hudsucker = "0.25" hudsucker = "0.25"

View file

@ -5,14 +5,14 @@ use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use base64::Engine; use base64::Engine;
use http::request::Request;
use http_body_util::BodyExt;
use hudsucker::certificate_authority::RcgenAuthority; use hudsucker::certificate_authority::RcgenAuthority;
use hudsucker::rcgen::{ use hudsucker::rcgen::{
BasicConstraints, CertificateParams, DnType, IsCa, Issuer, KeyPair, KeyUsagePurpose, BasicConstraints, CertificateParams, DnType, IsCa, Issuer, KeyPair, KeyUsagePurpose,
PKCS_ECDSA_P256_SHA256, PKCS_ECDSA_P256_SHA256,
}; };
use hudsucker::{Body, HttpContext, HttpHandler, Proxy, RequestOrResponse}; use hudsucker::{Body, HttpContext, HttpHandler, Proxy, RequestOrResponse};
use http::request::Request;
use http_body_util::BodyExt;
use md5::{Digest, Md5}; use md5::{Digest, Md5};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -74,11 +74,7 @@ impl Default for CaptureConfig {
impl Default for DeviceConfig { impl Default for DeviceConfig {
fn default() -> Self { fn default() -> Self {
Self { Self { serial: "127.0.0.1:5555".to_string(), wait_secs: 180, shutdown_bluestacks: true }
serial: "127.0.0.1:5555".to_string(),
wait_secs: 180,
shutdown_bluestacks: true,
}
} }
} }
@ -100,17 +96,13 @@ impl Default for ProxyConfig {
impl Default for CaptureSettings { impl Default for CaptureSettings {
fn default() -> Self { fn default() -> Self {
Self { Self { timeout_secs: 300, replay_timeout_secs: 30 }
timeout_secs: 300,
replay_timeout_secs: 30,
}
} }
} }
fn detect_adb() -> String { fn detect_adb() -> String {
if let Ok(local_app_data) = std::env::var("LOCALAPPDATA") { if let Ok(local_app_data) = std::env::var("LOCALAPPDATA") {
let candidate = let candidate = format!("{local_app_data}\\Android\\Sdk\\platform-tools\\adb.exe");
format!("{local_app_data}\\Android\\Sdk\\platform-tools\\adb.exe");
if Path::new(&candidate).exists() { if Path::new(&candidate).exists() {
return candidate; return candidate;
} }
@ -206,13 +198,11 @@ impl HttpHandler for CaptureHandler {
&& host.contains("topdrives") && host.contains("topdrives")
&& path_and_query.contains("/sync") && path_and_query.contains("/sync")
{ {
let body_bytes = BodyExt::collect(std::mem::replace( let body_bytes =
request.body_mut(), BodyExt::collect(std::mem::replace(request.body_mut(), Body::from(Vec::new())))
Body::from(Vec::new()), .await
)) .map(|collected| collected.to_bytes().to_vec())
.await .unwrap_or_default();
.map(|collected| collected.to_bytes().to_vec())
.unwrap_or_default();
let template = CapturedRequest { let template = CapturedRequest {
method: request.method().to_string(), method: request.method().to_string(),
@ -259,15 +249,13 @@ impl HttpHandler for CaptureHandler {
return response; return response;
} }
let bytes = match BodyExt::collect(std::mem::replace( let bytes =
response.body_mut(), match BodyExt::collect(std::mem::replace(response.body_mut(), Body::from(Vec::new())))
Body::from(Vec::new()), .await
)) {
.await Ok(collected) => collected.to_bytes().to_vec(),
{ Err(_) => return response,
Ok(collected) => collected.to_bytes().to_vec(), };
Err(_) => return response,
};
if contains_marker(&bytes, GARAGE_MARKER) { if contains_marker(&bytes, GARAGE_MARKER) {
if let Ok(json) = serde_json::from_slice::<serde_json::Value>(&bytes) { if let Ok(json) = serde_json::from_slice::<serde_json::Value>(&bytes) {
@ -321,7 +309,12 @@ async fn fetch_response(
let lower = name.as_str().to_lowercase(); let lower = name.as_str().to_lowercase();
if matches!( if matches!(
lower.as_str(), lower.as_str(),
"host" | "content-length" | "accept-encoding" | "connection" | "proxy-connection" | "keep-alive" "host"
| "content-length"
| "accept-encoding"
| "connection"
| "proxy-connection"
| "keep-alive"
) { ) {
continue; continue;
} }
@ -347,11 +340,10 @@ fn persist_capture(
json: &serde_json::Value, json: &serde_json::Value,
template: Option<&CapturedRequest>, template: Option<&CapturedRequest>,
) -> Result<()> { ) -> Result<()> {
let timestamp = time::OffsetDateTime::now_utc().format( let timestamp =
&time::format_description::parse_borrowed::<2>( time::OffsetDateTime::now_utc().format(&time::format_description::parse_borrowed::<2>(
"[year]-[month]-[day]T[hour]-[minute]-[second]", "[year]-[month]-[day]T[hour]-[minute]-[second]",
)?, )?)?;
)?;
let run_dir = project_root.join("captures").join(&timestamp); let run_dir = project_root.join("captures").join(&timestamp);
std::fs::create_dir_all(&run_dir)?; std::fs::create_dir_all(&run_dir)?;
@ -561,19 +553,14 @@ fn inject_ca(config: &CaptureConfig, cert_pem: &str, project_root: &Path) -> Res
let local_path = project_root.join("ca").join("topdrives-ca.pem"); let local_path = project_root.join("ca").join("topdrives-ca.pem");
std::fs::write(&local_path, cert_pem)?; std::fs::write(&local_path, cert_pem)?;
let presence = adb_shell( let presence =
config, adb_shell(config, &format!("test -f {store_path} && echo present || echo absent"))?;
&format!("test -f {store_path} && echo present || echo absent"),
)?;
if presence.trim() == "present" { if presence.trim() == "present" {
println!("CA already present in the system trust store"); println!("CA already present in the system trust store");
return Ok(()); return Ok(());
} }
adb( adb(config, &["push", local_path.to_string_lossy().as_ref(), staging_path])?;
config,
&["push", local_path.to_string_lossy().as_ref(), staging_path],
)?;
adb_shell_root( adb_shell_root(
config, config,
"mkdir -p /data/local/tmp/cacerts-backup && cp /system/etc/security/cacerts/* /data/local/tmp/cacerts-backup/", "mkdir -p /data/local/tmp/cacerts-backup && cp /system/etc/security/cacerts/* /data/local/tmp/cacerts-backup/",
@ -619,13 +606,15 @@ fn setup_adb_reverse(config: &CaptureConfig, port: u16) -> Result<()> {
Err(_) => { Err(_) => {
let route = adb_shell(config, "ip route")?; let route = adb_shell(config, "ip route")?;
let gateway = route.lines().find_map(|line| { let gateway = route.lines().find_map(|line| {
line.strip_prefix("default via ") line.strip_prefix("default via ").and_then(|rest| rest.split(' ').next())
.and_then(|rest| rest.split(' ').next())
}); });
match gateway { match gateway {
Some(gateway) => { Some(gateway) => {
println!("adb reverse failed, falling back to gateway address {gateway}:{port}"); println!(
adb_shell(config, &format!("settings put global http_proxy {gateway}:{port}")).map(|_| ()) "adb reverse failed, falling back to gateway address {gateway}:{port}"
);
adb_shell(config, &format!("settings put global http_proxy {gateway}:{port}"))
.map(|_| ())
} }
None => Err("could not set up adb reverse and no default gateway found".into()), None => Err("could not set up adb reverse and no default gateway found".into()),
} }
@ -638,15 +627,7 @@ fn launch_game(config: &CaptureConfig) -> Result<()> {
std::thread::sleep(Duration::from_secs(2)); std::thread::sleep(Duration::from_secs(2));
adb( adb(
config, config,
&[ &["shell", "monkey", "-p", GAME_PACKAGE, "-c", "android.intent.category.LAUNCHER", "1"],
"shell",
"monkey",
"-p",
GAME_PACKAGE,
"-c",
"android.intent.category.LAUNCHER",
"1",
],
)?; )?;
Ok(()) Ok(())
} }
@ -671,8 +652,9 @@ fn build_proxy(
Ok(Box::pin(proxy.start())) Ok(Box::pin(proxy.start()))
} }
type ProxyFuture = type ProxyFuture = std::pin::Pin<
std::pin::Pin<Box<dyn std::future::Future<Output = std::result::Result<(), hudsucker::Error>> + Send>>; Box<dyn std::future::Future<Output = std::result::Result<(), hudsucker::Error>> + Send>,
>;
async fn command_run(config: &CaptureConfig) -> Result<()> { async fn command_run(config: &CaptureConfig) -> Result<()> {
let project_root = std::env::current_dir()?; let project_root = std::env::current_dir()?;
@ -704,15 +686,13 @@ async fn command_run(config: &CaptureConfig) -> Result<()> {
output: output_path(config, &project_root), output: output_path(config, &project_root),
pending: std::sync::Mutex::new(None), pending: std::sync::Mutex::new(None),
}); });
let proxy = build_proxy(listener, &cert_pem, &key_pem, CaptureHandler { state: state.clone() })?; let proxy =
build_proxy(listener, &cert_pem, &key_pem, CaptureHandler { state: state.clone() })?;
tokio::spawn(proxy); tokio::spawn(proxy);
set_device_proxy(config, config.proxy.port)?; set_device_proxy(config, config.proxy.port)?;
setup_adb_reverse(config, config.proxy.port)?; setup_adb_reverse(config, config.proxy.port)?;
println!( println!("proxy active on 127.0.0.1:{}, launching {GAME_PACKAGE} ...", config.proxy.port);
"proxy active on 127.0.0.1:{}, launching {GAME_PACKAGE} ...",
config.proxy.port
);
launch_game(config)?; launch_game(config)?;
println!( println!(
@ -770,7 +750,12 @@ async fn command_replay(config: &CaptureConfig) -> Result<()> {
let lower = name.to_lowercase(); let lower = name.to_lowercase();
if matches!( if matches!(
lower.as_str(), lower.as_str(),
"host" | "content-length" | "accept-encoding" | "connection" | "proxy-connection" | "keep-alive" "host"
| "content-length"
| "accept-encoding"
| "connection"
| "proxy-connection"
| "keep-alive"
) { ) {
continue; continue;
} }
@ -799,12 +784,7 @@ async fn command_replay(config: &CaptureConfig) -> Result<()> {
); );
} }
let json: serde_json::Value = serde_json::from_slice(&bytes)?; let json: serde_json::Value = serde_json::from_slice(&bytes)?;
persist_capture( persist_capture(&project_root, &output_path(config, &project_root), &json, Some(&template))?;
&project_root,
&output_path(config, &project_root),
&json,
Some(&template),
)?;
println!("replay finished successfully"); println!("replay finished successfully");
Ok(()) Ok(())
} }
@ -899,9 +879,7 @@ mod tests {
let filename = android_ca_filename(CA_COMMON_NAME); let filename = android_ca_filename(CA_COMMON_NAME);
assert_eq!(filename.len(), 10); assert_eq!(filename.len(), 10);
assert!(filename.ends_with(".0")); assert!(filename.ends_with(".0"));
assert!(filename[..8] assert!(filename[..8].chars().all(|character| character.is_ascii_hexdigit()));
.chars()
.all(|character| character.is_ascii_hexdigit()));
} }
#[test] #[test]

View file

@ -11,6 +11,7 @@ use std::sync::Arc;
use dashmap::DashMap; use dashmap::DashMap;
use futures::future::join_all; use futures::future::join_all;
use indicatif::{ProgressBar, ProgressStyle};
use serde_json::Value; use serde_json::Value;
use tokio::sync::Semaphore; use tokio::sync::Semaphore;
use tokio::task; use tokio::task;
@ -363,25 +364,43 @@ const CAR_TIMING_API_BASE: &str = "https://api.topdrivesrecords.com/car/";
/// Fetches timing data files for all cars using controlled concurrency. /// Fetches timing data files for all cars using controlled concurrency.
/// ///
/// Displays a single-line progress counter (completed/total cars) that stays
/// on the bottom line while downloads run. The counter tracks completions,
/// not cars seen, since downloads run concurrently under the semaphore.
///
/// Returns the first error encountered, or `Ok(())` if all downloads succeeded. /// Returns the first error encountered, or `Ok(())` if all downloads succeeded.
pub async fn fetch_times(cars_db_path: &str, times_dir: &str) -> Result<(), AppError> { pub async fn fetch_times(cars_db_path: &str, times_dir: &str) -> Result<(), AppError> {
let car_list = get_car_list(cars_db_path)?;
let progress = ProgressBar::new(car_list.len() as u64);
progress.set_style(
ProgressStyle::with_template("{msg} {pos}/{len}")
.expect("static progress template is valid"),
);
progress.set_message("Downloading track times for car");
let semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT)); let semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT));
let mut handles = vec![]; let mut handles = vec![];
for car in get_car_list(cars_db_path)? { for car in car_list {
let permit = semaphore let permit = semaphore
.clone() .clone()
.acquire_owned() .acquire_owned()
.await .await
.map_err(|e| AppError::Network(format!("Semaphore closed: {e}")))?; .map_err(|e| AppError::Network(format!("Semaphore closed: {e}")))?;
let times_dir = times_dir.to_string(); let times_dir = times_dir.to_string();
let progress = progress.clone();
handles.push(tokio::spawn(async move { handles.push(tokio::spawn(async move {
let _permit = permit; // held until end of block let _permit = permit; // held until end of block
fetch_car(&car.rid, &times_dir).await let result = fetch_car(&car.rid, &times_dir).await;
progress.inc(1);
result
})); }));
} }
let results = join_all(handles).await; let results = join_all(handles).await;
// Clear the counter before error logging so warnings don't garble the line.
progress.finish_and_clear();
let mut first_error: Option<AppError> = None; let mut first_error: Option<AppError> = None;
for result in results { for result in results {
match result { match result {