Stream Epic chunks straight into staged files with a budgeted shared cache

The native adaptive install: chunks fetched in parallel through the fetch
core in first-use order; single-use chunks are written at their part's
offset into the file's staging file, shared chunks are kept in a budgeted
cache (furthest-next-use eviction, refetch later), and every file is
SHA-1 checked, synced and atomically renamed with the bounded writer's
rules. New JNI export nativeStream.

app/src/main/rust/stores/JNI.md

@@ -112,7 +112,12 @@ On-disk protocol (what a resumed run, by either engine, relies on): a file is as
 
 ## `com.droiddeck.launcher.stores.epic.EpicNative`
 
-Two runs make up the native half of an Epic install:
+The manager's default Epic install is the **stream** run (`nativeStream`, below): the chunks
+are fetched in parallel and written straight into staged files as they arrive. The older batch
+pair below (fetch everything into a cache, then assemble) stays for A/B benchmarks (the hidden
+`storesEpicMode=batch` preference).
+
+Two runs make up the native half of a batch Epic install:
 
 1. **Fetch** (`nativeStart`): fills the chunk cache with verified, decompressed chunks for the
    files the manager says are pending. The Kotlin manager still fetches and parses the manifest
@@ -224,6 +229,50 @@ Bannerlator's `BlEpicDownload.run(...)` wraps `nativeStart` in a blocking call t
 `onComplete`, then `nativeRelease` in `finally`. The same wrapper shape fits `nativeAssemble`
 (cancel is honoured between files; a multi-GB file finishes before the run stops).
 
+### `nativeStream` — the adaptive install
+
+| Method | JNI signature | Symbol |
+|---|---|---|
+| `nativeStream(...)`: Long | `([BLjava/lang/String;Ljava/lang/String;[Ljava/lang/String;[IJJLjava/lang/String;IILjava/lang/Object;)J` (last parameter is the listener) | `Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeStream` |
+
+```kotlin
+@JvmStatic private external fun nativeStream(
+    manifest: ByteArray, installDir: String, chunkCacheDir: String, cdnPrefixes: Array<String>,
+    pendingFileIdx: IntArray, expectedBytes: Long, cacheBudget: Long, caBundlePath: String,
+    maxWorkers: Int, processWorkers: Int, listener: StreamListener,
+): Long
+```
+
+Parameters as `nativeStart`, except `expectedBytes` (the manager's `Σ fileSize` of the pending
+files, `-1` = no cross-check; no chunk count) and `cacheBudget` (the most bytes of shared chunks
+the cache may hold; the manager sizes it from free space, `0` = never cache, refetch instead).
+`chunkCacheDir` is `""` (`<installDir>/.chunks`) or a scratch directory; the manager falls back to
+`""` for a card install when internal storage cannot hold the budget.
+
+What the run does (`epic/stream.rs`): a chunk one file part uses is decompressed, verified and
+written at the part's offset into that file's unique sibling `.droiddeck-epic-*.stage` file; a
+chunk several parts use is also kept in the cache, decompressed, as `<GUID>` (a later copy after
+an eviction as `<GUID>.<n>`) and deleted after its last covered part. The plan evicts by furthest
+next use when the budget would be exceeded and fetches the evicted chunk again later. Each file,
+once complete: synced, size and SHA-1 checked, `onBeforeReplace()` once for the whole run, then
+renamed over its destination. Cancel or failure removes every unfinished stage file before
+`onComplete`, leaves the destinations and the cache; success removes the cache directory.
+
+Planning runs on the calling thread and also prepares the cache (keeps verified chunks this
+install needs, deletes everything else, as the Java writer does). A refusal returns `0` after a
+synchronous `onComplete(false, reason, 0)`; nothing was fetched or written, and the manager runs
+its bounded Java writer.
+
+| `StreamListener` method | JNI signature | When |
+|---|---|---|
+| `onStreamProgress(bytesWritten: Long, bytesTotal: Long, filesDone: Int, filesTotal: Int, filesBytes: Long)` | `(JJIIJ)V` | after written parts (at most every 50 ms) and after every finished file; `filesBytes` = sizes of the files swapped in |
+| `onLog(line: String)` | `(Ljava/lang/String;)V` | engine lines: `stream-plan …`, the fetch core's lines, `stream-summary mode=adaptive …` (peak cache, peak staging, refetches, direct vs cached bytes) |
+| `onBeforeReplace(): Boolean` | `()Z` | once, before the first file replaces its destination; `false` or a throw fails the run with nothing replaced |
+| `onComplete(success: Boolean, error: String, bytes: Long)` | `(ZLjava/lang/String;J)V` | terminal, exactly once; `bytes` = file bytes written; `error` = `"cancelled"` on cancel |
+
+Unlike the batch runs, the Kotlin wrapper (`EpicNative.stream`) waits for `onComplete` after a
+cancel without a time limit: the stage-file cleanup must be finished before the manager returns.
+
 ## `com.droiddeck.launcher.stores.amazon.AmazonNative`
 
 The file-fetch pool of an Amazon Games install: one streamed HTTP GET per manifest file, SHA-256

app/src/main/rust/stores/src/epic.rs

@@ -18,6 +18,9 @@
 //! - [`assemble`] — the files from the cache, one sequential write each, SHA-1 checked as they
 //!   are written; with a scratch cache (one the caller placed apart from the install) each chunk
 //!   is deleted as soon as the last file that uses it is done.
+//! - [`stream`] — the adaptive install the manager uses by default: the chunks fetched in
+//!   parallel through the fetch core and written straight into staging files as they arrive,
+//!   shared chunks kept in a budgeted cache, each file verified and swapped in safely.
 //! - [`jni`] — `Java_com_droiddeck_launcher_stores_epic_EpicNative_native*` exports.
 //!
 //! The selective-install-tag, delta/resume and chunk verification rules mirrored here were ported
@@ -31,6 +34,7 @@ pub mod driver;
 pub mod jni;
 pub mod manifest;
 pub mod plan;
+pub mod stream;
 
 /// User-Agent the Java manager sends on every chunk request (`EpicDownloadManager.UA`).
 pub const USER_AGENT: &str =

app/src/main/rust/stores/src/epic/chunk.rs

@@ -177,6 +177,80 @@ fn write_part(
     Ok(written)
 }
 
+/// The window a chunk decompresses to: the manifest's, or 1 MiB for a legacy JSON manifest that
+/// names none (`EpicDownloadManager.chunkWindowSize`). The streaming installer also plans its
+/// shared-chunk cache with this size.
+pub fn chunk_window_bytes(window_size: i32) -> u64 {
+    if window_size > 0 {
+        window_size as u64
+    } else {
+        1024 * 1024
+    }
+}
+
+/// One downloaded body → the verified, decompressed chunk in memory, for the streaming installer
+/// (`stream.rs`), which writes most chunks straight into the files that use them and never puts
+/// them in a cache. The checks are the bounded Java writer's (`downloadChunkStreaming`), which is
+/// stricter than [`write_verified_chunk`]: a stored payload shorter than `compressedSize` is
+/// truncated, a zlib stream must reach its end, the output may not grow past the chunk's window
+/// and must fill it exactly when the manifest names one, and the SHA-1 must match when the
+/// manifest has one. `Err` = "try another CDN", exactly like a cache write that failed.
+pub fn decode_verified_chunk(
+    body: &[u8],
+    window_size: i32,
+    expected_sha1: Option<&[u8; 20]>,
+) -> Result<Vec<u8>, String> {
+    let hdr = parse_chunk_header(body)?;
+    let start = hdr.payload_start();
+    if start > body.len() {
+        return Err("Stream ended during skip".to_string());
+    }
+    let want = if hdr.compressed_size <= 0 {
+        0
+    } else {
+        hdr.compressed_size as usize
+    };
+    let avail = body.len() - start;
+    let cap = chunk_window_bytes(window_size) as usize;
+    let out = if hdr.is_compressed() {
+        // Java stops reading once the inflater finishes, so a body may carry less than
+        // `compressedSize` as long as the stream itself is complete.
+        let payload = &body[start..start + want.min(avail)];
+        let mut out: Vec<u8> = Vec::with_capacity(cap + 1);
+        let mut inflater = flate2::Decompress::new(true);
+        let status = inflater
+            .decompress_vec(payload, &mut out, flate2::FlushDecompress::Finish)
+            .map_err(|e| format!("inflate: {e}"))?;
+        if out.len() > cap {
+            return Err("chunk exceeds expected window size".to_string());
+        }
+        if status != flate2::Status::StreamEnd {
+            return Err("truncated compressed Epic chunk".to_string());
+        }
+        out
+    } else {
+        if avail < want {
+            return Err("truncated Epic chunk".to_string());
+        }
+        if want > cap {
+            return Err("chunk exceeds expected window size".to_string());
+        }
+        body[start..start + want].to_vec()
+    };
+    if window_size > 0 && out.len() != window_size as usize {
+        return Err(format!(
+            "chunk is {} bytes, window is {window_size}",
+            out.len()
+        ));
+    }
+    if let Some(expected) = expected_sha1 {
+        if Sha1::digest(&out).as_slice() != &expected[..] {
+            return Err("Chunk SHA-1 mismatch (streaming)".to_string());
+        }
+    }
+    Ok(out)
+}
+
 #[cfg(test)]
 pub(crate) mod test_support {
     use super::*;
@@ -361,6 +435,29 @@ mod tests {
         let _ = fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    fn decode_in_memory_applies_the_bounded_writer_checks() {
+        let data = sample_data();
+        let window = data.len() as i32;
+        let sha = sha1_of(&data);
+        let zipped = build_chunk_body(&data, true, 2, None);
+        assert_eq!(decode_verified_chunk(&zipped, window, Some(&sha)).unwrap(), data);
+        let stored = build_chunk_body(&data, false, 0, None);
+        assert_eq!(decode_verified_chunk(&stored, window, Some(&sha)).unwrap(), data);
+        // No window and no hash (legacy JSON manifest): accepted up to 1 MiB.
+        assert_eq!(decode_verified_chunk(&zipped, 0, None).unwrap(), data);
+        // Window mismatch, SHA-1 mismatch, truncation, oversize.
+        assert!(decode_verified_chunk(&zipped, window + 1, None).is_err());
+        let err = decode_verified_chunk(&zipped, window, Some(&[7; 20])).unwrap_err();
+        assert!(err.contains("SHA-1 mismatch"), "{err}");
+        let cut = zipped.len() - 500;
+        assert!(decode_verified_chunk(&zipped[..cut], window, None).is_err());
+        assert!(decode_verified_chunk(&stored[..stored.len() - 1], 0, None)
+            .unwrap_err()
+            .contains("truncated"));
+        assert!(decode_verified_chunk(&zipped, window - 1, None).is_err());
+    }
+
     #[test]
     fn stale_part_file_is_cleared_before_the_attempt() {
         let dir = temp_dir("stale");

app/src/main/rust/stores/src/epic/jni.rs

@@ -7,7 +7,11 @@
 //!             listener: EpicNative.Listener) -> long handle (0 = not started)
 //! nativeAssemble(manifest: byte[], installDir: String, chunkCacheDir: String, fileIdx: int[],
 //!             listener: EpicNative.Listener) -> long handle (0 = not started)
-//! nativeCancel(handle)      — flips the run's AtomicBool; fetch or assembly stops promptly
+//! nativeStream(manifest: byte[], installDir: String, chunkCacheDir: String, cdnPrefixes: String[],
+//!             pendingFileIdx: int[], expectedBytes: long, cacheBudget: long, caBundlePath: String,
+//!             maxWorkers: int, processWorkers: int, listener: EpicNative.StreamListener)
+//!             -> long handle (0 = not started)
+//! nativeCancel(handle)      — flips the run's AtomicBool; fetch, assembly or stream stops promptly
 //! nativeRelease(handle)     — frees the handle (after onComplete)
 //! ```
 //! `chunkCacheDir` = "" keeps the cache at `<installDir>/.chunks`; any other directory is a
@@ -17,6 +21,11 @@
 //! an assembly run, per file written), `onLog(String line)`,
 //! `onComplete(boolean success, String error, long bytes)`.
 //!
+//! The stream run (`stream.rs`) writes the files as the chunks arrive; its listener has
+//! `onStreamProgress(long bytesWritten, long bytesTotal, int filesDone, int filesTotal,
+//! long filesBytes)`, `onLog(String)`, `boolean onBeforeReplace()` (called once, before the first
+//! file replaces its destination; `false` fails the run) and the same `onComplete`.
+//!
 //! Planning (manifest parse + cross-check) runs synchronously on the calling thread so a plan
 //! failure returns 0 BEFORE any fetch and Java can fall back to its own pool; the fetch runs on
 //! a new thread. JavaVM attach / GlobalRef handling is the same shape as the GOG facade.
@@ -31,6 +40,7 @@ use jni::{JNIEnv, JavaVM};
 
 use super::assemble::{build_assemble_plan, run_assemble, AssembleRequest};
 use super::driver::{build_plan, run_plan, EpicRequest};
+use super::stream::{build_stream_plan, run_stream, Hooks, StreamProgress, StreamRequest};
 
 #[cfg(target_os = "android")]
 #[link(name = "log")]
@@ -458,6 +468,184 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
     Box::into_raw(handle) as jlong
 }
 
+fn call_on_stream_progress(env: &mut JNIEnv, listener: &JObject, p: &StreamProgress) {
+    if listener.is_null() {
+        return;
+    }
+    let _ = env.call_method(
+        listener,
+        "onStreamProgress",
+        "(JJIIJ)V",
+        &[
+            JValue::Long(p.bytes_written as jlong),
+            JValue::Long(p.bytes_total as jlong),
+            JValue::Int(p.files_done.min(i32::MAX as u64) as jint),
+            JValue::Int(p.files_total.min(i32::MAX as u64) as jint),
+            JValue::Long(p.files_bytes as jlong),
+        ],
+    );
+    clear_pending_exception(env);
+}
+
+/// `boolean onBeforeReplace()`: the Java side marks an existing install `updating`. A thrown
+/// exception or `false` is a refusal.
+fn call_on_before_replace(env: &mut JNIEnv, listener: &JObject) -> Result<(), String> {
+    if listener.is_null() {
+        return Ok(());
+    }
+    let result = env
+        .call_method(listener, "onBeforeReplace", "()Z", &[])
+        .and_then(|v| v.z());
+    let threw = env.exception_check().unwrap_or(false);
+    clear_pending_exception(env);
+    match result {
+        Ok(true) if !threw => Ok(()),
+        _ => Err("could not mark the install as updating".to_string()),
+    }
+}
+
+/// `EpicNative.nativeStream(...)`: the adaptive install (see `stream.rs`). Parse, plan and cache
+/// preparation run on the calling thread, so a refusal returns 0 (after a synchronous
+/// `onComplete(false, reason, 0)`) before anything is fetched or written, and the manager runs
+/// the bounded Java writer instead. The run itself is on a new thread; `onComplete` fires after
+/// every unfinished staging file of a cancelled or failed run is gone.
+#[no_mangle]
+pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeStream(
+    mut env: JNIEnv,
+    _class: JClass,
+    manifest: JByteArray,
+    install_dir: JString,
+    chunk_cache_dir: JString,
+    cdn_prefixes: JObjectArray,
+    pending_file_idx: JIntArray,
+    expected_bytes: jlong,
+    cache_budget: jlong,
+    ca_bundle_path: JString,
+    max_workers: jint,
+    process_workers: jint,
+    listener: JObject,
+) -> jlong {
+    if listener.is_null() {
+        android_log("nativeStream: null listener");
+        return 0;
+    }
+    let Ok(manifest_bytes) = env.convert_byte_array(manifest) else {
+        clear_pending_exception(&mut env);
+        call_on_complete(&mut env, &listener, false, "manifest bytes unreadable", 0);
+        return 0;
+    };
+    let install_dir = jstring_to_string(&mut env, &install_dir).unwrap_or_default();
+    let chunk_cache_dir = jstring_to_string(&mut env, &chunk_cache_dir).unwrap_or_default();
+    let cdn_prefixes = string_array_to_vec(&mut env, &cdn_prefixes);
+    let pending_file_indices: Vec<usize> = int_array_to_vec(&env, &pending_file_idx)
+        .into_iter()
+        .map(|i| i.max(0) as usize)
+        .collect();
+    let ca_bundle_path = jstring_to_string(&mut env, &ca_bundle_path).unwrap_or_default();
+    let max_workers = max_workers.max(1) as usize;
+    let process_workers = process_workers.max(1) as usize;
+    let label = {
+        let base = std::path::Path::new(&install_dir)
+            .file_name()
+            .map(|s| s.to_string_lossy().into_owned())
+            .unwrap_or_default();
+        format!("epic app={base}")
+    };
+    log_both(
+        &mut env,
+        &listener,
+        &format!(
+            "engine=rust-stream label=\"{label}\" install_dir={install_dir} cache_dir={} cdns={} pending_files={} cache_budget={} workers={max_workers} process_workers={process_workers} manifest_bytes={}",
+            if chunk_cache_dir.is_empty() { "<installDir>/.chunks" } else { chunk_cache_dir.as_str() },
+            cdn_prefixes.len(),
+            pending_file_indices.len(),
+            cache_budget.max(0),
+            manifest_bytes.len()
+        ),
+    );
+
+    let req = StreamRequest {
+        manifest_bytes,
+        install_dir,
+        chunk_cache_dir,
+        cdn_prefixes,
+        pending_file_indices,
+        expected_bytes: if expected_bytes >= 0 {
+            Some(expected_bytes as u64)
+        } else {
+            None
+        },
+        cache_budget: cache_budget.max(0) as u64,
+        ca_bundle_path,
+        max_workers,
+        process_workers,
+        label,
+    };
+    let plan = match build_stream_plan(&req) {
+        Ok(plan) => plan,
+        Err(err) => {
+            log_both(&mut env, &listener, &format!("not started: {err}"));
+            call_on_complete(&mut env, &listener, false, &err, 0);
+            return 0;
+        }
+    };
+    let Ok(vm) = env.get_java_vm() else {
+        clear_pending_exception(&mut env);
+        call_on_complete(&mut env, &listener, false, "JavaVM unavailable", 0);
+        return 0;
+    };
+    let Ok(listener) = env.new_global_ref(&listener) else {
+        clear_pending_exception(&mut env);
+        call_on_complete(&mut env, &listener, false, "listener ref failed", 0);
+        return 0;
+    };
+
+    let cancel = Arc::new(AtomicBool::new(false));
+    let thread_cancel = Arc::clone(&cancel);
+    let handle = Box::new(EpicDownloadHandle { cancel });
+
+    thread::spawn(move || {
+        crate::priority::background();
+        let progress = |p: &StreamProgress| {
+            let Ok(mut env) = vm.attach_current_thread_as_daemon() else {
+                return;
+            };
+            call_on_stream_progress(&mut env, listener.as_obj(), p);
+        };
+        let before_replace = || {
+            let Ok(mut env) = vm.attach_current_thread_as_daemon() else {
+                return Err("JavaVM attach failed".to_string());
+            };
+            call_on_before_replace(&mut env, listener.as_obj())
+        };
+        let log = |line: &str| {
+            android_log(line);
+            let Ok(mut env) = vm.attach_current_thread_as_daemon() else {
+                return;
+            };
+            call_on_log(&mut env, listener.as_obj(), line);
+        };
+        let hooks = Hooks {
+            progress: &progress,
+            before_replace: &before_replace,
+            log: &log,
+        };
+        let outcome = run_stream(&plan, &req, thread_cancel.as_ref(), &hooks);
+        let Ok(mut env) = vm.attach_current_thread_as_daemon() else {
+            return;
+        };
+        call_on_complete(
+            &mut env,
+            listener.as_obj(),
+            outcome.success,
+            &outcome.error,
+            outcome.bytes_written,
+        );
+    });
+
+    Box::into_raw(handle) as jlong
+}
+
 #[no_mangle]
 pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeCancel(
     _env: JNIEnv,

tools/release/check_android_libraries.py

@@ -14,6 +14,7 @@ EXPORTS = ['Java_com_droiddeck_launcher_stores_StoresNative_nativeVersion',
            'Java_com_droiddeck_launcher_stores_gog_GogNative_nativeStart',
            'Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeStart',
            'Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeAssemble',
+           'Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeStream',
            'Java_com_droiddeck_launcher_stores_amazon_AmazonNative_nativeStart']