diff --git a/app/src/main/rust/stores/JNI.md b/app/src/main/rust/stores/JNI.md index 447088741..9f1b38550 100644 --- a/app/src/main/rust/stores/JNI.md +++ b/app/src/main/rust/stores/JNI.md @@ -70,7 +70,7 @@ after it (marker files, exe resolution, DLC markers, redist orchestration). | Method | JNI signature | Symbol | |---|---|---| | `nativeProbe(): Int` | `()I` | `Java_com_droiddeck_launcher_stores_gog_GogNative_nativeProbe` | -| `nativeStart(...)`: Long | `(I[Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;[Ljava/lang/String;Ljava/lang/String;IIZLjava/lang/String;Ljava/lang/Object;)J` (last parameter is the listener; its static type is the app's choice) | `Java_com_droiddeck_launcher_stores_gog_GogNative_nativeStart` | +| `nativeStart(...)`: Long | `(I[Ljava/lang/String;[Ljava/lang/String;Ljava/lang/String;[Ljava/lang/String;Ljava/lang/String;IIZLjava/lang/String;Ljava/lang/Object;)J` (last parameter is the listener; its static type is the app's choice) | `Java_com_droiddeck_launcher_stores_gog_GogNative_nativeStart` | | `nativeCancel(handle: Long)` | `(J)V` | `Java_com_droiddeck_launcher_stores_gog_GogNative_nativeCancel` | | `nativeRelease(handle: Long)` | `(J)V` | `Java_com_droiddeck_launcher_stores_gog_GogNative_nativeRelease` | @@ -83,7 +83,7 @@ once at startup instead of mid-download. |---|---|---| | 1 | `kind: Int` | `0` = `KIND_GEN2_CHUNKS`, `1` = `KIND_GEN1_RANGES` (anything else → gen2) | | 2 | `depotManifests: Array` | gen2: the inflated depot-manifest JSON strings, in fetch order, already filtered by product + language; gen1: the inflated build manifest (one element) | -| 3 | `cdnBase: String` | gen2: the resolved secure-link base (query string intact) or the unauthenticated dependency store base; gen1: `""` (file URLs are in the manifest) | +| 3 | `cdnBases: Array` | gen2: **every** CDN base of the secure-link answer — one per `urls[]` entry, each resolved the way `parseCdnUrl` resolves `urls[0]` today (query string intact), in GOG's order — or the one unauthenticated dependency store base (an array of one); gen1: `emptyArray()` (file URLs are in the manifest). Blank entries are ignored and a base whose host already appeared is dropped (the first wins). The fetch core spreads the window across the distinct hosts with a per-host cap of `max(6, ceil(maxWorkers / hosts))`, rotating a failed chunk to another CDN; one base behaves exactly like the old single `cdnBase`. The start log line lists them as `hosts=a,b,c`. | | 4 | `installDir: String` | absolute install directory | | 5 | `skipPaths: Array` | relative paths already completed by an earlier run of this same download (secure-link refresh re-run): counted done without re-hashing, no progress event | | 6 | `caBundlePath: String` | see conventions | @@ -94,14 +94,15 @@ once at startup instead of mid-download. | 11 | `listener` | see below | Returns `0` without any callback when: the listener is null, the JavaVM cannot be obtained, the -install dir or manifest list is empty, `cdnBase` is empty for gen2, or the worker thread cannot be -spawned. Otherwise the run is on a thread named `gog-dl` and `onComplete` fires exactly once. +install dir or manifest list is empty, `cdnBases` holds no usable base for gen2, or the worker +thread cannot be spawned. Otherwise the run is on a thread named `gog-dl` and `onComplete` fires exactly once. Suggested listener: `com.droiddeck.launcher.stores.gog.GogNative.Listener` (an interface). | Listener method | JNI signature | When | |---|---|---| | `onProgress(bytesDone: Long, bytesTotal: Long, filesDone: Int, filesTotal: Int, file: String, fileBytes: Long, verified: Boolean)` | `(JJIILjava/lang/String;JZ)V` | one file reached its final state. `verified = true` = resume-skip (existing file passed size+MD5; no bytes credited, `fileBytes` = its size); `false` = freshly assembled, size+MD5-verified and renamed (`fileBytes` = its decompressed size). `filesDone` counts both, `bytesDone` counts assembled bytes only. Process-pool / worker thread. | +| `onBytes(bytesDone: Long, bytesTotal: Long)` | `(JJ)V` | byte progress between file completions: every 250 ms while the run is fetching, only when the value moved, plus once when the fetch ends. `bytesDone` = bytes of assembled files + bytes of files still in flight (gen2: their finished chunks, decompressed; gen1: what the current attempt has written; a retried attempt starts over, so nothing is counted twice); `bytesTotal` is the same total `onProgress` reports. It never goes past the next `onProgress.bytesDone` by more than the in-flight files and is equal to it once they finish. Drive the progress bar and speed from this; keep `onProgress` for file counts and per-file events. Called from the engine's ticker thread. Optional: a listener without it still works (the call fails quietly). | | `onLog(line: String)` | `(Ljava/lang/String;)V` | engine diagnostics; the same line already went to logcat under the tag `GogNative`. Any native thread. | | `onComplete(success: Boolean, cancelled: Boolean, linkExpiry: Boolean, error: String, bytesWritten: Long, filesDone: Int)` | `(ZZZLjava/lang/String;JI)V` | exactly once, from the worker thread, last. `linkExpiry = true` = the run died on an HTTP 401/403/404/500, the codes the GOG secure link returns once it expired: refresh the link and call `nativeStart` again with the finished files in `skipPaths`. `error` is `""` on success. | diff --git a/app/src/main/rust/stores/src/gog/engine.rs b/app/src/main/rust/stores/src/gog/engine.rs index 42bb42a50..0233110c3 100644 --- a/app/src/main/rust/stores/src/gog/engine.rs +++ b/app/src/main/rust/stores/src/gog/engine.rs @@ -39,7 +39,8 @@ pub const TMP_SUFFIX: &str = ".bhtmp"; /// Floor for the per-host cap (the core's default); a single-host store gets the whole ceiling. pub const PER_HOST_CAP_FLOOR: usize = 6; -/// `max(6, ceil(max_workers / distinct_hosts))` — GOG's one CDN host gets the whole ceiling. +/// `max(6, ceil(max_workers / distinct_hosts))` — one CDN host gets the whole ceiling, several +/// share it. pub fn per_host_cap_for(max_workers: usize, distinct_hosts: usize) -> usize { let hosts = distinct_hosts.max(1); let workers = max_workers.max(1); @@ -74,9 +75,12 @@ pub struct GogRequest { /// gen2: inflated depot-manifest JSON strings, in the order Java fetched them (already filtered /// by base-product / DLC-product and language in Java). gen1: the inflated build manifest. pub depot_manifests: Vec, - /// gen2: resolved CDN base from `parseCdnUrl` (secure-link query string kept), or the - /// unauthenticated dependency store base. gen1: unused (file URLs live in the manifest). - pub cdn_base: String, + /// gen2: the resolved CDN bases from the secure-link answer (every `urls[]` entry, query + /// strings kept, in the order GOG lists them), or the one unauthenticated dependency store + /// base. Several bases on distinct hosts let the fetch core spread the window across CDNs + /// with its per-host cap; a base on a host already listed is dropped (the first one wins). + /// gen1: unused (file URLs live in the manifest). + pub cdn_bases: Vec, pub install_dir: String, /// Files already completed by an earlier run of this same download (secure-link refresh /// re-run): counted as done WITHOUT re-hashing and WITHOUT a progress event. @@ -109,6 +113,10 @@ pub trait GogEvents: Sync { bytes_total: u64, ); fn on_log(&self, line: &str); + /// Periodic byte progress while files are in flight (~4/s, only when the value moved): + /// assembled bytes plus the bytes of files not yet finished. Same `bytes_total` as + /// `on_file_done`. Default: ignored (test recorders). + fn on_bytes(&self, _bytes_done: u64, _bytes_total: u64) {} } #[derive(Clone, Debug, Default, PartialEq, Eq)] @@ -178,6 +186,9 @@ pub fn is_link_expiry_status(code: u16) -> bool { struct FileState { handle: Option>, chunks_done: usize, + /// Decompressed bytes of this file's finished chunks, credited to the sink's `partial` + /// counter until the file itself is finalized and moves into `bytes_done`. + partial: u64, finalized: bool, failed: bool, tmp_path: PathBuf, @@ -322,6 +333,9 @@ struct GogSink<'a> { bytes_done: &'a AtomicU64, events: &'a dyn GogEvents, speed: Mutex, + /// Bytes of files still in flight (finished chunks; for gen1, the current attempt's pieces). + /// `bytes_done + partial` is what the byte ticker reports between file completions. + partial: AtomicU64, } impl<'a> GogSink<'a> { @@ -374,7 +388,7 @@ impl<'a> GogSink<'a> { /// One verified chunk landed: count it; on the file's last chunk run the whole-file verify + /// rename and fire the per-file progress event (Java: doneCount++, totalBytes += df.totalSize, /// "Downloading: "). - fn chunk_done(&self, file_idx: usize) -> Result<(), SinkError> { + fn chunk_done(&self, file_idx: usize, chunk_bytes: u64) -> Result<(), SinkError> { let file = &self.files[file_idx]; let mut st = self .states @@ -388,6 +402,8 @@ impl<'a> GogSink<'a> { ))); } st.chunks_done += 1; + st.partial += chunk_bytes; + self.partial.fetch_add(chunk_bytes, Ordering::Relaxed); if st.chunks_done < file.chunks.len() { return Ok(()); } @@ -395,12 +411,14 @@ impl<'a> GogSink<'a> { st.failed = true; return Err(SinkError::Fatal(err)); } + let partial = std::mem::take(&mut st.partial); drop(st); let done = self.files_done.fetch_add(1, Ordering::Relaxed) + 1; let bytes = self .bytes_done .fetch_add(file.total_size, Ordering::Relaxed) + file.total_size; + self.partial.fetch_sub(partial, Ordering::Relaxed); self.events.on_file_done( &file.relative_path, file.total_size, @@ -541,7 +559,7 @@ impl<'a> FetchSink for GogSink<'a> { } drop(handle); let credited = inflated.len() as u64; - self.chunk_done(file_idx)?; + self.chunk_done(file_idx, credited)?; Ok(credited) } @@ -694,11 +712,59 @@ impl<'a> FetchSink for GogSink<'a> { return Err(SinkError::Retry(msg)); } drop(writer.file); - self.chunk_done(file_idx)?; + self.chunk_done(file_idx, writer.written)?; Ok(0) } } +/// How often the byte ticker reports while files are in flight. +pub const BYTE_TICK: Duration = Duration::from_millis(250); + +/// The CDN bases to fetch from: empty entries dropped, then one per distinct host (`host_key`), +/// the first base seen for a host kept, order preserved. +pub fn distinct_bases(bases: &[String]) -> Vec { + let mut seen: HashSet = HashSet::new(); + bases + .iter() + .map(|b| b.trim()) + .filter(|b| !b.is_empty() && seen.insert(host_key(b))) + .map(str::to_string) + .collect() +} + +/// Runs `fetch` with a reporter thread beside it that calls `on_bytes(bytes_done + partial, +/// bytes_total)` every [`BYTE_TICK`] when the value moved. `on_file_done` only fires when a whole +/// file is finished, so a game made of a few large files gave the UI a progress bar that sat +/// still and then jumped; this fills the gaps. One last report is made when `fetch` returns. +fn with_byte_ticker( + events: &dyn GogEvents, + bytes_done: &AtomicU64, + partial: &AtomicU64, + bytes_total: u64, + fetch: impl FnOnce() -> T, +) -> T { + let stop = AtomicBool::new(false); + let current = || bytes_done.load(Ordering::Relaxed) + partial.load(Ordering::Relaxed); + std::thread::scope(|scope| { + scope.spawn(|| { + crate::priority::background(); + let mut last = u64::MAX; + while !stop.load(Ordering::Relaxed) { + std::thread::sleep(BYTE_TICK); + let now = current(); + if now != last { + last = now; + events.on_bytes(now, bytes_total); + } + } + }); + let out = fetch(); + stop.store(true, Ordering::Relaxed); + events.on_bytes(current(), bytes_total); + out + }) +} + /// Runs one download loop. Blocking; returns when every pending file is finalized, on the first /// fatal file failure, or on cancel. Safe to call again with `skip_paths` extended by the files /// this run reported done (that is how the Java side re-runs after a secure-link refresh). @@ -718,11 +784,12 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go let chunk_count: usize = files.iter().map(|f| f.chunks.len()).sum(); let max_workers = req.max_workers.max(1); let process_workers = req.process_workers.max(1); - let host = host_key(&req.cdn_base); - let per_host_cap = per_host_cap_for(max_workers, 1); + let bases = distinct_bases(&req.cdn_bases); + let hosts: Vec = bases.iter().map(|b| host_key(b)).collect(); + let per_host_cap = per_host_cap_for(max_workers, hosts.len()); events.on_log(&format!( - "engine=rust kind=gen2 mode=stream label={} files={} chunks={} planned_bytes={} skip_paths={} workers={} per_host_cap={} process_workers={} sort_largest_first={} host={} {}", + "engine=rust kind=gen2 mode=stream label={} files={} chunks={} planned_bytes={} skip_paths={} workers={} per_host_cap={} process_workers={} sort_largest_first={} hosts={} {}", req.label, files_total, chunk_count, @@ -732,7 +799,7 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go per_host_cap, process_workers, req.sort_largest_first, - host, + hosts.join(","), crate::priority::log_field() )); @@ -816,6 +883,7 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go states.push(Mutex::new(FileState { handle: None, chunks_done: 0, + partial: 0, // Files not pending (verified or skip-listed) are already final for this run. finalized: !is_pending, failed: false, @@ -828,10 +896,11 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go pending_files += 1; pending_bytes = pending_bytes.saturating_add(file.total_size); for (chunk_idx, chunk) in file.chunks.iter().enumerate() { - let url = build_chunk_url(&req.cdn_base, &build_cdn_path(&chunk.hash)); + let path = build_cdn_path(&chunk.hash); items.push(FetchItem { id: table.len() as u64, - urls: vec![url], + // One URL per host, in `hosts` order: the core picks `urls[host_idx]`. + urls: bases.iter().map(|b| build_chunk_url(b, &path)).collect(), reserve: chunk.compressed_size, range: None, }); @@ -870,11 +939,11 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go bytes_done: &bytes_done, events, speed: Mutex::new(SpeedSampler::new()), + partial: AtomicU64::new(0), }; - let hosts = vec![host]; let opts = FetchOptions { max_workers, - // One CDN host gets the whole ceiling (Java has no per-host cap either). + // The ceiling split across the distinct CDN hosts (floor 6); one host gets all of it. per_host_cap, timeout: CHUNK_TIMEOUT, headers: vec![("User-Agent".to_string(), USER_AGENT.to_string())], @@ -887,7 +956,9 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go }; let log = |line: &str| events.on_log(line); let progress = |_bytes: u64, _items_ok: u64| {}; - let outcome = run_fetch(items, &hosts, &opts, &sink, cancel, &progress, &log); + let outcome = with_byte_ticker(events, &bytes_done, &sink.partial, planned_bytes, || { + run_fetch(items, &hosts, &opts, &sink, cancel, &progress, &log) + }); let cancelled = outcome.cancelled || cancel.load(Ordering::Relaxed); let mut error = outcome.error.clone().unwrap_or_default(); @@ -945,6 +1016,9 @@ fn run_gen2(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go struct Gen1State { handle: Option, + /// Bytes this file's current attempt has written (credited to the sink's `partial`); a new + /// attempt starts over, so a retry never counts its bytes twice. + attempt_bytes: u64, out_path: PathBuf, finalized: bool, } @@ -962,6 +1036,9 @@ struct Gen1Sink<'a> { bytes_done: &'a AtomicU64, events: &'a dyn GogEvents, speed: Mutex, + /// Bytes of files still in flight (finished chunks; for gen1, the current attempt's pieces). + /// `bytes_done + partial` is what the byte ticker reports between file completions. + partial: AtomicU64, } impl<'a> FetchSink for Gen1Sink<'a> { @@ -985,6 +1062,8 @@ impl<'a> FetchSink for Gen1Sink<'a> { .map_err(|_| SinkError::Fatal("gen1 state poisoned".to_string()))?; if offset == 0 { // New attempt: Java `outFile.delete()` + `new FileOutputStream(out)`. + self.partial + .fetch_sub(std::mem::take(&mut st.attempt_bytes), Ordering::Relaxed); st.handle = None; if let Some(parent) = st.out_path.parent() { let _ = fs::create_dir_all(parent); @@ -1016,6 +1095,8 @@ impl<'a> FetchSink for Gen1Sink<'a> { st.handle = None; return Err(SinkError::Retry(msg)); } + st.attempt_bytes += data.len() as u64; + self.partial.fetch_add(data.len() as u64, Ordering::Relaxed); Ok(()) } @@ -1030,9 +1111,12 @@ impl<'a> FetchSink for Gen1Sink<'a> { .map_err(|_| SinkError::Fatal("gen1 state poisoned".to_string()))?; st.handle = None; // close st.finalized = true; + let attempt = std::mem::take(&mut st.attempt_bytes); + self.bytes_done.fetch_add(file.size, Ordering::Relaxed); + self.partial.fetch_sub(attempt, Ordering::Relaxed); } let done = self.files_done.fetch_add(1, Ordering::Relaxed) + 1; - let bytes = self.bytes_done.fetch_add(file.size, Ordering::Relaxed) + file.size; + let bytes = self.bytes_done.load(Ordering::Relaxed); self.events.on_file_done( &file.path, file.size, @@ -1115,6 +1199,7 @@ fn run_gen1(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go .unwrap_or(false); states.push(Mutex::new(Gen1State { handle: None, + attempt_bytes: 0, out_path, finalized: skipped || resumed, })); @@ -1169,6 +1254,7 @@ fn run_gen1(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go bytes_done: &bytes_done, events, speed: Mutex::new(SpeedSampler::new()), + partial: AtomicU64::new(0), }; let hosts = vec![host]; let opts = FetchOptions { @@ -1183,7 +1269,9 @@ fn run_gen1(req: &GogRequest, cancel: &AtomicBool, events: &dyn GogEvents) -> Go }; let log = |line: &str| events.on_log(line); let progress = |_bytes: u64, _items_ok: u64| {}; - let outcome = run_fetch(items, &hosts, &opts, &sink, cancel, &progress, &log); + let outcome = with_byte_ticker(events, &bytes_done, &sink.partial, planned_bytes, || { + run_fetch(items, &hosts, &opts, &sink, cancel, &progress, &log) + }); let cancelled = outcome.cancelled || cancel.load(Ordering::Relaxed); let error = outcome.error.clone().unwrap_or_default(); @@ -1286,6 +1374,24 @@ mod tests { assert_eq!(http_status_in("HTTP 404 not found"), Some(404)); assert_eq!(http_status_in("chunk=abc403def timed out"), None, "hash digits are not a status"); assert_eq!(http_status_in("timeout"), None); + let bases = vec![ + "https://gog-cdn-fastly.gog.com/token/a?x=1".to_string(), + " ".to_string(), + "https://GOG-CDN-FASTLY.gog.com/token/b?x=2".to_string(), + "https://gog-cdn-akamai.gog.com/token/c?y=3".to_string(), + "https://gog-cdn-lumen.gog.com/token/d".to_string(), + ]; + assert_eq!( + distinct_bases(&bases), + vec![ + "https://gog-cdn-fastly.gog.com/token/a?x=1".to_string(), + "https://gog-cdn-akamai.gog.com/token/c?y=3".to_string(), + "https://gog-cdn-lumen.gog.com/token/d".to_string(), + ], + "blank dropped, second fastly base dropped, order kept" + ); + assert_eq!(per_host_cap_for(96, 3), 32); + assert_eq!(per_host_cap_for(96, 1), 96); assert!(is_link_expiry_status(401)); assert!(is_link_expiry_status(403)); assert!(is_link_expiry_status(404)); @@ -1345,6 +1451,7 @@ mod tests { states: vec![Mutex::new(FileState { handle: None, chunks_done: 0, + partial: 0, finalized: false, failed: false, tmp_path: tmp_path.clone(), @@ -1357,6 +1464,7 @@ mod tests { bytes_done: &bytes_done, events: &rec, speed: Mutex::new(SpeedSampler::new()), + partial: AtomicU64::new(0), }; let item = |id: u64| FetchItem { id, urls: vec!["x".into()], reserve: 0, range: None }; @@ -1401,6 +1509,7 @@ mod tests { states: vec![Mutex::new(FileState { handle: None, chunks_done: 0, + partial: 0, finalized: false, failed: false, tmp_path: tmp_path.clone(), @@ -1413,6 +1522,7 @@ mod tests { bytes_done: &bytes_done, events: &rec, speed: Mutex::new(SpeedSampler::new()), + partial: AtomicU64::new(0), }; let item = FetchItem { id: 0, urls: vec!["x".into()], reserve: 0, range: None }; match sink.process(&item, raw) { @@ -1438,7 +1548,7 @@ mod tests { ); let req = GogRequest { depot_manifests: vec![manifest], - cdn_base: "https://example.invalid/store?tok=1".to_string(), + cdn_bases: vec!["https://example.invalid/store?tok=1".to_string()], install_dir: dir.to_string_lossy().into_owned(), skip_paths: vec!["skipped.bin".to_string()], max_workers: 4, @@ -1505,6 +1615,7 @@ mod tests { states: vec![Mutex::new(FileState { handle: None, chunks_done: 0, + partial: 0, finalized: false, failed: false, tmp_path: tmp_path.clone(), @@ -1517,6 +1628,7 @@ mod tests { bytes_done: &bytes_done, events: &rec, speed: Mutex::new(SpeedSampler::new()), + partial: AtomicU64::new(0), }; let item = |id: u64| FetchItem { id, urls: vec!["x".into()], reserve: 0, range: None }; @@ -1580,6 +1692,7 @@ mod tests { files: &files, states: vec![Mutex::new(Gen1State { handle: None, + attempt_bytes: 0, out_path: dir.join("sub/blob.bin"), finalized: false, })], @@ -1589,6 +1702,7 @@ mod tests { bytes_done: &bytes_done, events: &rec, speed: Mutex::new(SpeedSampler::new()), + partial: AtomicU64::new(0), }; let item = FetchItem { id: 0, urls: vec!["x".into()], reserve: 8, range: Some((100, 107)) }; // Attempt 1 writes a partial, attempt 2 restarts from offset 0 and truncates it. @@ -1642,7 +1756,7 @@ mod tests { let manifest = r#"{"depot":{"items":[{"path":"a.bin","chunks":[{"compressedMd5":"ab","size":4}]}]}}"#; let req = GogRequest { depot_manifests: vec![manifest.to_string()], - cdn_base: "https://example.invalid/store".to_string(), + cdn_bases: vec!["https://example.invalid/store".to_string()], install_dir: dir.to_string_lossy().into_owned(), max_workers: 1, process_workers: 1, @@ -1655,4 +1769,32 @@ mod tests { assert_eq!(res.error, "cancelled"); let _ = fs::remove_dir_all(&dir); } + + #[test] + fn byte_ticker_reports_in_flight_bytes_between_file_completions() { + struct Bytes(StdMutex>); + impl GogEvents for Bytes { + #[allow(clippy::too_many_arguments)] + fn on_file_done(&self, _: &str, _: u64, _: bool, _: u32, _: u32, _: u64, _: u64) {} + fn on_log(&self, _: &str) {} + fn on_bytes(&self, done: u64, total: u64) { + self.0.lock().unwrap().push((done, total)); + } + } + let rec = Bytes(StdMutex::new(Vec::new())); + let bytes_done = AtomicU64::new(100); + let partial = AtomicU64::new(0); + let out = with_byte_ticker(&rec, &bytes_done, &partial, 1000, || { + partial.store(50, Ordering::Relaxed); + std::thread::sleep(BYTE_TICK * 3); + bytes_done.store(400, Ordering::Relaxed); + partial.store(0, Ordering::Relaxed); + 7 + }); + assert_eq!(out, 7); + let calls = rec.0.lock().unwrap().clone(); + assert!(calls.contains(&(150, 1000)), "in-flight bytes reported mid-run: {calls:?}"); + assert_eq!(calls.last(), Some(&(400, 1000)), "final report when the fetch returns"); + assert!(calls.windows(2).all(|w| w[0] != w[1] || w[1] == (400, 1000)), "no repeats while idle: {calls:?}"); + } } diff --git a/app/src/main/rust/stores/src/gog/jni.rs b/app/src/main/rust/stores/src/gog/jni.rs index 9d3920d0c..eee602a6b 100644 --- a/app/src/main/rust/stores/src/gog/jni.rs +++ b/app/src/main/rust/stores/src/gog/jni.rs @@ -136,6 +136,23 @@ impl GogEvents for JniEvents { }); } + fn on_bytes(&self, bytes_done: u64, bytes_total: u64) { + // A listener without `onBytes` (an app built against the older contract) makes this a + // NoSuchMethodError, which clear_exception drops: the run is unaffected. + self.with_env(|env| { + let _ = env.call_method( + self.listener.as_obj(), + "onBytes", + "(JJ)V", + &[ + JValue::Long(bytes_done as jlong), + JValue::Long(bytes_total as jlong), + ], + ); + Self::clear_exception(env); + }); + } + fn on_log(&self, line: &str) { android_log(line); self.with_env(|env| { @@ -211,7 +228,7 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_gog_GogNative_nativeSt _class: JClass, kind: jint, depot_manifests: JObjectArray, - cdn_base: JString, + cdn_bases: JObjectArray, install_dir: JString, skip_paths: JObjectArray, ca_bundle_path: JString, @@ -236,7 +253,7 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_gog_GogNative_nativeSt let request = GogRequest { kind: PlanKind::from_i32(kind), depot_manifests: string_array_to_vec(&mut env, &depot_manifests), - cdn_base: jstring_to_string(&mut env, &cdn_base).unwrap_or_default(), + cdn_bases: string_array_to_vec(&mut env, &cdn_bases), install_dir: jstring_to_string(&mut env, &install_dir).unwrap_or_default(), skip_paths: string_array_to_vec(&mut env, &skip_paths), ca_bundle_path: jstring_to_string(&mut env, &ca_bundle_path).unwrap_or_default(), @@ -246,11 +263,11 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_gog_GogNative_nativeSt label: jstring_to_string(&mut env, &label).unwrap_or_else(|| "gog".to_string()), }; let needs_base = request.kind == PlanKind::Gen2Chunks; - if (needs_base && request.cdn_base.is_empty()) + if (needs_base && super::engine::distinct_bases(&request.cdn_bases).is_empty()) || request.install_dir.is_empty() || request.depot_manifests.is_empty() { - android_log("nativeStart: invalid request (empty cdn base / install dir / manifests)"); + android_log("nativeStart: invalid request (no cdn base / empty install dir / no manifests)"); return 0; }