refactor: modular architecture with unified error handling and DashMap concurrency

- Break monolithic records.rs into commands/, output/, parsing/ modules
- Add lib.rs with shared types (AppError, Car, BestCarRecord, etc.)
- Replace Arc<Mutex<HashMap>> with DashMap for concurrent access
- Add unified AppError enum with thiserror
- Add configurable CLI paths (--my-cars, --cars-db, --times-dir, --tracks-db)
- Add serde deserialization for Car struct
- Add CSV escaping for fields with commas/quotes/newlines
- Fix track name splitting: split_once -> rsplit_once for _a handling
- Add integration tests, proptest property-based tests, and unit tests
- Add rustfmt.toml configuration
- Update documentation (API, ARCHITECTURE, DATA_STRUCTURES, README)
This commit is contained in:
Jan-Peter von Hunnius 2026-05-08 16:19:01 +02:00
parent 9d57b18a4e
commit a9a46ee551
24 changed files with 3541 additions and 1470 deletions

652
src/parsing/mod.rs Normal file
View file

@ -0,0 +1,652 @@
//! Core parsing logic for car timing data.
//!
//! Handles per-car JSON timing file parsing, best-car record updates,
//! and orchestrated concurrent parsing of all timing files.
pub mod car_list;
use std::collections::HashSet;
use std::fs;
use std::sync::Arc;
use dashmap::DashMap;
use futures::future::join_all;
use serde_json::Value;
use tokio::sync::Semaphore;
use tokio::task;
use top_drives::{
AppError, AnalysisContext, BestCarRecord, Car, TripletKey,
translate_track_name, translate_track_type,
};
use self::car_list::get_car_list;
// ---------------------------------------------------------------------------
// Constants (shared across parsing modules)
// ---------------------------------------------------------------------------
/// Maximum RQ level in the Top Drives database.
/// Used as the upper bound when iterating RQ ranges and as a fallback for parse errors.
pub const MAX_RQ: u8 = 113;
/// Minimum valid track time in seconds. Times below this threshold are discarded.
pub const MIN_VALID_TIME: f64 = 1.2;
/// Tolerance for considering two times as equal (5ms).
pub const TIME_TOLERANCE: f64 = 0.005;
/// Multiplier constant for Test Bowl time inversion.
/// Test Bowl tracks use speed rather than time, so we invert: `FACTOR / time`.
pub const TEST_BOWL_FACTOR: f64 = 2000.0;
/// Maximum number of concurrent download/parse tasks.
///
/// IMPORTANT: Keep ≤ 50 to avoid exhausting tokio's default blocking thread pool (512 threads),
/// since each concurrent permit holder also occupies a `spawn_blocking` thread.
pub const CONCURRENCY_LIMIT: usize = 20;
// ---------------------------------------------------------------------------
// Core parsing: per-car timing data
// ---------------------------------------------------------------------------
/// Parses timing data for a single car from its JSON file.
///
/// **Blocking** — performs synchronous file I/O and CPU-intensive JSON parsing.
/// Must be called from [`tokio::task::spawn_blocking`], never directly from an async context,
/// as it would block the tokio runtime thread.
///
/// This is the core function responsible for processing individual car performance data.
/// It reads the car's JSON file, extracts timing information for each track and condition,
/// filters out invalid or downvoted times, and updates the analysis context.
///
/// # Arguments
///
/// * `car` - The car data structure containing metadata about the car to process
/// * `my_cars` - Set of GUIDs the user owns (used to mark ownership)
/// * `best_cars` - Shared concurrent map of best car records (`DashMap`)
/// * `times_dir` - Directory containing per-car timing JSON files
pub fn parse_car_times(
car: Car,
my_cars: &HashSet<String>,
best_cars: &DashMap<TripletKey, Vec<BestCarRecord>>,
times_dir: &str,
) -> Result<(), AppError> {
// Construct file path for this car's timing data
let file_name = car.rid.to_owned() + ".json";
let local_file_path = format!("{times_dir}/{file_name}");
// Read file content
let file_content = match fs::read_to_string(&local_file_path) {
Ok(content) => content,
Err(e) => {
// File not found is expected for some cars — treat as non-fatal
eprintln!("Warning: could not read {local_file_path}: {e}");
return Ok(());
}
};
// Parse JSON data
let json_data: Value = match serde_json::from_str(&file_content) {
Ok(v) => v,
Err(e) => {
return Err(AppError::Parse {
file: local_file_path,
reason: e.to_string(),
});
}
};
// Process each configuration variant (111, 233, 323, etc.)
if let Some(configurations) = json_data["data"].as_object() {
for (config, entry) in configurations {
// Process timing data for each track
if let Some(times) = entry["times"].as_object() {
for (track_id, details) in times {
// Parse track ID to separate base track name from surface condition.
// Use rsplit_once to split on the LAST "_a" so track names that
// contain "_a" (e.g. "airplaneHangars") are handled correctly.
if let Some((cleaned_track_id, track_type)) = track_id.rsplit_once("_a") {
let mut time = details["t"].as_f64().unwrap_or(0.0);
let down_votes = details["down"].as_array();
let no_down_votes = match down_votes {
None => true,
Some(votes) => votes.is_empty(),
};
if track_id.starts_with("testBowl") {
if time > 0.0 {
time = TEST_BOWL_FACTOR / time;
} else {
continue; // skip invalid zero time
}
}
let owned = my_cars.contains(&car.guid);
if time > MIN_VALID_TIME && no_down_votes {
let track_name = translate_track_name(cleaned_track_id);
let surface = translate_track_type(track_type);
update_best_car(
best_cars,
&track_name,
car.rq,
surface,
BestCarRecord {
car_rid: car.rid.clone(),
brand: car.brand.clone(),
only_name: car.only_name.clone(),
year: car.year,
rq: car.rq,
config: config.clone(),
time,
owned,
prize: car.prize,
guid: car.guid.clone(),
},
);
}
}
}
}
}
}
Ok(())
}
// ---------------------------------------------------------------------------
// Core parsing: update best car records
// ---------------------------------------------------------------------------
/// Updates the best cars database with a new performance record.
///
/// Implements the core logic for determining the best performing cars across different
/// RQ ranges. For each RQ level from the car's base RQ up to MAX_RQ, this
/// function determines whether the new record represents an improvement or tie.
///
/// # Allocation Optimization
///
/// The record is cloned at most once (for the first insertion), and subsequent
/// RQ levels reuse the entry via `DashMap`'s `entry()` API, avoiding per-RQ
/// clone overhead for the common "insert new" case.
///
/// String keys (`track_id`, `surface`) are pre-allocated once and cloned per
/// iteration, avoiding repeated `to_string()` formatting overhead.
pub fn update_best_car(
best_cars: &DashMap<TripletKey, Vec<BestCarRecord>>,
track_id: &str,
car_rq: u8,
surface: &str,
new_record: BestCarRecord,
) {
let track_id_owned = track_id.to_string();
let surface_owned = surface.to_string();
for rq in car_rq..=MAX_RQ {
let is_last = rq == MAX_RQ;
let key = (track_id_owned.clone(), rq, surface_owned.clone());
match best_cars.entry(key) {
dashmap::mapref::entry::Entry::Occupied(mut entry) => {
let existing_list = entry.get_mut();
let best_time = existing_list.first().map(|r| r.time).unwrap_or(f64::MAX);
if new_record.time < best_time {
if is_last {
*existing_list = vec![new_record];
return;
} else {
*existing_list = vec![new_record.clone()];
}
} else if (new_record.time - best_time).abs() < TIME_TOLERANCE
&& !existing_list
.iter()
.any(|r| r.car_rid == new_record.car_rid)
{
if is_last {
existing_list.push(new_record);
return;
} else {
existing_list.push(new_record.clone());
}
}
// else: record is discarded for this RQ — no clone needed
}
dashmap::mapref::entry::Entry::Vacant(entry) => {
if is_last {
entry.insert(vec![new_record]);
return;
} else {
entry.insert(vec![new_record.clone()]);
}
}
}
}
}
// ---------------------------------------------------------------------------
// Orchestrated parsing
// ---------------------------------------------------------------------------
/// Orchestrates concurrent parsing of all car timing files.
///
/// Spawns a separate async task for each car in the database. The semaphore is
/// acquired in the async context (avoiding `block_on` inside `spawn_blocking`,
/// which risks deadlocking the blocking thread pool), then the synchronous
/// parsing work runs inside `spawn_blocking` while the permit is held.
///
/// Takes `Arc<HashSet<String>>` to avoid cloning the entire owned-car set on
/// every call.
pub async fn parse_all_times(
my_cars: Arc<HashSet<String>>,
ctx: &AnalysisContext,
cars_db_path: &str,
times_dir: &str,
) -> Result<Vec<Car>, AppError> {
let car_list = get_car_list(cars_db_path)?;
let mut handles = vec![];
let semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT));
let best_cars = Arc::clone(&ctx.best_cars);
for car in car_list.iter() {
let my_cars = Arc::clone(&my_cars);
let best_cars = Arc::clone(&best_cars);
let car = car.clone();
let sem = Arc::clone(&semaphore);
let times_dir = times_dir.to_string();
let handle = task::spawn(async move {
// Acquire the semaphore asynchronously — propagate error if closed.
let _permit = match sem.acquire().await {
Ok(p) => p,
Err(_) => return Err(AppError::Internal("Semaphore closed".into())),
};
// Run the synchronous parsing on a blocking thread while holding the permit.
task::spawn_blocking(move || {
parse_car_times(car, &my_cars, &best_cars, &times_dir)
}).await.unwrap_or_else(|e| Err(AppError::Internal(format!("Blocking task join error: {e}"))))
});
handles.push(handle);
}
let results = join_all(handles).await;
let mut parse_errors = 0usize;
for result in results {
match result {
// Outer JoinError from tokio::spawn
Err(e) => {
eprintln!("Warning: parsing task failed: {e}");
parse_errors += 1;
}
// parse_car_times or semaphore returned an error
Ok(Err(e)) => {
eprintln!("Warning: {e}");
parse_errors += 1;
}
// parse_car_times succeeded
Ok(Ok(())) => {}
}
}
if parse_errors > 0 {
eprintln!("Warning: {parse_errors} car(s) had parse errors during analysis");
}
// If every car failed to parse, the data is unusable — propagate the error.
if parse_errors == car_list.len() && !car_list.is_empty() {
return Err(AppError::Parse {
file: times_dir.to_string(),
reason: format!("All {parse_errors} car(s) failed to parse — data directory may be corrupted"),
});
}
Ok(car_list)
}
// ---------------------------------------------------------------------------
// Data fetching
// ---------------------------------------------------------------------------
/// Remote URL for the per-car timing API.
const CAR_TIMING_API_BASE: &str = "https://api.topdrivesrecords.com/car/";
/// Fetches timing data files for all cars using controlled concurrency.
///
/// 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> {
let semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT));
let mut handles = vec![];
for car in get_car_list(cars_db_path)? {
let permit = semaphore.clone().acquire_owned().await
.map_err(|e| AppError::Network(format!("Semaphore closed: {e}")))?;
let times_dir = times_dir.to_string();
handles.push(tokio::spawn(async move {
let _permit = permit; // held until end of block
fetch_car(&car.rid, &times_dir).await
}));
}
let results = join_all(handles).await;
let mut first_error: Option<AppError> = None;
for result in results {
match result {
Ok(Ok(())) => {}
Ok(Err(e)) => {
eprintln!("{e}");
if first_error.is_none() { first_error = Some(e); }
}
Err(e) => {
let msg = AppError::Network(format!("Task join error: {e}"));
eprintln!("{msg}");
if first_error.is_none() { first_error = Some(msg); }
}
}
}
first_error.map_or(Ok(()), Err)
}
/// Downloads timing data for a single car from the remote API.
async fn fetch_car(rid: &str, times_dir: &str) -> Result<(), AppError> {
let file_name = rid.to_owned() + ".json";
let url = format!("{CAR_TIMING_API_BASE}{rid}");
let local_file_path = format!("{times_dir}/{file_name}");
println!("Fetching {url}.");
top_drives::get_file(&url, &local_file_path).await
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashSet;
// -- Helper: create a test BestCarRecord --
fn make_record(car_rid: &str, rq: u8, time: f64, guid: &str) -> BestCarRecord {
BestCarRecord {
car_rid: car_rid.to_string(),
brand: "TestBrand".to_string(),
only_name: "TestModel".to_string(),
year: 2020,
rq,
config: "111".to_string(),
time,
owned: false,
prize: false,
guid: guid.to_string(),
}
}
// =======================================================================
// update_best_car tests
// =======================================================================
#[test]
fn test_update_best_car_inserts_new_record() {
let ctx = AnalysisContext::new();
let record = make_record("Car_A", 85, 100.0, "guid_a");
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record);
let key = ("laguna".to_string(), 85, "Dry".to_string());
assert!(ctx.best_cars.contains_key(&key));
let records = ctx.best_cars.get(&key).unwrap();
assert_eq!(records.len(), 1);
assert_eq!(records[0].car_rid, "Car_A");
}
#[test]
fn test_update_best_car_replaces_slower_record() {
let ctx = AnalysisContext::new();
let slow = make_record("Car_A", 85, 110.0, "guid_a");
let fast = make_record("Car_B", 85, 95.0, "guid_b");
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", slow);
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", fast);
let records = ctx.best_cars.get(&("laguna".to_string(), 85, "Dry".to_string())).unwrap();
assert_eq!(records.len(), 1);
assert_eq!(records[0].car_rid, "Car_B"); // faster car wins
}
#[test]
fn test_update_best_car_tie_within_tolerance() {
let ctx = AnalysisContext::new();
let record1 = make_record("Car_A", 85, 100.0, "guid_a");
let record2 = make_record("Car_B", 85, 100.003, "guid_b"); // within 0.005
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record1);
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record2);
let records = ctx.best_cars.get(&("laguna".to_string(), 85, "Dry".to_string())).unwrap();
assert_eq!(records.len(), 2, "Tied records should both be kept");
assert!(records.iter().any(|r| r.car_rid == "Car_A"));
assert!(records.iter().any(|r| r.car_rid == "Car_B"));
}
#[test]
fn test_update_best_car_outside_tolerance_replaces() {
let ctx = AnalysisContext::new();
let record1 = make_record("Car_A", 85, 100.0, "guid_a");
let record3 = make_record("Car_C", 85, 100.006, "guid_c"); // outside 0.005
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record1);
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record3);
let records = ctx.best_cars.get(&("laguna".to_string(), 85, "Dry".to_string())).unwrap();
assert_eq!(records.len(), 1, "Record outside tolerance should not tie");
assert_eq!(records[0].car_rid, "Car_A"); // original kept (same best time)
}
#[test]
fn test_update_best_car_rq_range_propagation() {
let ctx = AnalysisContext::new();
let record = make_record("Car_A", 85, 100.0, "guid_a");
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record);
for rq in 85..=MAX_RQ {
let key = ("laguna".to_string(), rq, "Dry".to_string());
assert!(
ctx.best_cars.contains_key(&key),
"Car should appear at RQ level {rq}"
);
}
// Should NOT appear below its RQ
let key_below = ("laguna".to_string(), 84, "Dry".to_string());
assert!(!ctx.best_cars.contains_key(&key_below));
}
#[test]
fn test_update_best_car_duplicate_rid_not_added_in_tie() {
let ctx = AnalysisContext::new();
let record1 = make_record("Car_A", 85, 100.0, "guid_a");
let record2 = make_record("Car_A", 85, 100.001, "guid_a"); // same car, within tolerance
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record1);
update_best_car(&ctx.best_cars, "laguna", 85, "Dry", record2);
let records = ctx.best_cars.get(&("laguna".to_string(), 85, "Dry".to_string())).unwrap();
assert_eq!(records.len(), 1, "Same car_rid should not appear twice in tie list");
}
// =======================================================================
// parse_car_times tests
// =======================================================================
#[test]
fn test_parse_car_times_valid_json() {
let ctx = AnalysisContext::new();
let car = Car {
rid: "TestCar_2020".to_string(),
brand: "TestBrand".to_string(),
only_name: "TestModel".to_string(),
year: 2020,
rq: 85,
prize: false,
guid: "guid_test".to_string(),
};
// Create a temporary directory and timing file
let dir = tempfile::tempdir().unwrap();
let times_dir = dir.path().to_str().unwrap();
let file_path = dir.path().join("TestCar_2020.json");
let json_content = r#"{
"data": {
"111": {
"times": {
"laguna_a00": { "t": 112.45, "u": "Player1" },
"mnCity_a01": { "t": 45.23, "u": "Player2" }
}
}
}
}"#;
fs::write(&file_path, json_content).unwrap();
let my_cars: HashSet<String> = HashSet::new();
parse_car_times(car, &my_cars, &ctx.best_cars, times_dir).unwrap();
// Should have entries for RQ 85..=113 for both tracks
assert!(ctx.best_cars.contains_key(&("Laguna Seca".to_string(), 85, "Dry".to_string())));
assert!(ctx.best_cars.contains_key(&("Monaco Streets".to_string(), 85, "Wet".to_string())));
}
#[test]
fn test_parse_car_times_filters_downvoted() {
let ctx = AnalysisContext::new();
let car = Car {
rid: "TestCar_2020".to_string(),
brand: "TestBrand".to_string(),
only_name: "TestModel".to_string(),
year: 2020,
rq: 85,
prize: false,
guid: "guid_test".to_string(),
};
let dir = tempfile::tempdir().unwrap();
let times_dir = dir.path().to_str().unwrap();
let file_path = dir.path().join("TestCar_2020.json");
let json_content = r#"{
"data": {
"111": {
"times": {
"laguna_a00": { "t": 112.45, "u": "Player1", "down": ["Player3"] }
}
}
}
}"#;
fs::write(&file_path, json_content).unwrap();
let my_cars: HashSet<String> = HashSet::new();
parse_car_times(car, &my_cars, &ctx.best_cars, times_dir).unwrap();
// Downvoted time should be filtered out
assert!(
!ctx.best_cars.contains_key(&("Laguna Seca".to_string(), 85, "Dry".to_string())),
"Downvoted times should be excluded"
);
}
#[test]
fn test_parse_car_times_test_bowl_inversion() {
let ctx = AnalysisContext::new();
let car = Car {
rid: "TestCar_2020".to_string(),
brand: "TestBrand".to_string(),
only_name: "TestModel".to_string(),
year: 2020,
rq: 85,
prize: false,
guid: "guid_test".to_string(),
};
let dir = tempfile::tempdir().unwrap();
let times_dir = dir.path().to_str().unwrap();
let file_path = dir.path().join("TestCar_2020.json");
let json_content = r#"{
"data": {
"111": {
"times": {
"testBowl_a00": { "t": 50.0, "u": "Player1" }
}
}
}
}"#;
fs::write(&file_path, json_content).unwrap();
let my_cars: HashSet<String> = HashSet::new();
parse_car_times(car, &my_cars, &ctx.best_cars, times_dir).unwrap();
// Test Bowl time should be inverted: 2000.0 / 50.0 = 40.0
let records = ctx
.best_cars
.get(&("Test Bowl".to_string(), 85, "Dry".to_string()))
.unwrap();
assert_eq!(records[0].time, 40.0);
}
#[test]
fn test_parse_car_times_missing_file() {
let ctx = AnalysisContext::new();
let car = Car {
rid: "NonExistent".to_string(),
brand: "TestBrand".to_string(),
only_name: "TestModel".to_string(),
year: 2020,
rq: 85,
prize: false,
guid: "guid_test".to_string(),
};
let my_cars: HashSet<String> = HashSet::new();
// Should not panic — just print a warning and return Ok(())
parse_car_times(car, &my_cars, &ctx.best_cars, "/nonexistent/times").unwrap();
assert_eq!(ctx.best_cars.len(), 0, "No entries for missing file");
}
// =======================================================================
// Track name / type tests (from lib)
// =======================================================================
#[test]
fn test_translate_track_name_known_tracks() {
assert_eq!(translate_track_name("laguna"), "Laguna Seca");
assert_eq!(translate_track_name("mnCity"), "Monaco Streets");
assert_eq!(translate_track_name("testBowl"), "Test Bowl");
assert_eq!(translate_track_name("unknown_track"), "unknown_track");
}
#[test]
fn test_translate_track_type_known_conditions() {
assert_eq!(translate_track_type("00"), "Dry");
assert_eq!(translate_track_type("01"), "Wet");
assert_eq!(translate_track_type("10"), "Dirt");
assert_eq!(translate_track_type("99"), "99");
}
// =======================================================================
// Constants tests
// =======================================================================
#[test]
fn test_constants_are_reasonable() {
assert_eq!(MAX_RQ, 113);
assert!(MIN_VALID_TIME > 0.0);
assert!(TIME_TOLERANCE > 0.0);
assert!(TEST_BOWL_FACTOR > 0.0);
assert!(CONCURRENCY_LIMIT > 0);
}
}