Add adaptive parallel Epic installation with safe staged recovery

app/src/main/java/com/droiddeck/launcher/stores/epic/EpicDownloadManager.java

@@ -6,6 +6,7 @@ import android.util.Log;
 import com.droiddeck.launcher.stores.StoreLog;
 import com.droiddeck.launcher.stores.StoreInstallRoot;
 import com.droiddeck.launcher.stores.StoreInstalls;
+import com.droiddeck.launcher.stores.StoresNative;
 
 import org.json.JSONArray;
 import org.json.JSONObject;
@@ -22,6 +23,8 @@ import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.security.MessageDigest;
 import java.util.ArrayList;
@@ -32,15 +35,21 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.zip.Inflater;
 
 /**
  * The Epic install pipeline, ported from Bannerlator's EpicDownloadManager: the manifest API's CDN
  * list, the manifest binary (ChunksV4, or the older JSON form), the files narrowed to the base
  * set plus the device's language, a delta pass that skips files already on disk with the right
- * size and SHA-1, then the bounded-cache chunk-to-file writer. The native batch engine remains
- * available for other callers, but this install path intentionally uses the same serial writer
- * whether or not native code is installed: a whole-game chunk cache can exceed available space.
+ * size and SHA-1, then native adaptive parallel fetch/staged assembly when the library loads.
+ * Builds without the library use bounded Java parallel prefetch and one staged writer instead.
+ * Neither production path requires a whole-game chunk cache.
  *
  * Chunk URLs carry no auth tokens; the subfolder is the decimal group number; the Cloudflare
  * mirror is skipped (it refuses chunks). Blocking; the download queue runs it on its own thread.
@@ -51,6 +60,9 @@ public final class EpicDownloadManager {
     private static final String UA = EpicAuthClient.USER_AGENT;
     /** Keep reusable chunk data bounded even when a manifest shares chunks across distant files. */
     static final long MAX_RETAINED_CHUNK_CACHE_BYTES = 64L * 1024 * 1024;
+    static final int PREFETCH_PARTS = 8;
+    static final String STAGING_DIRECTORY = ".droiddeck-epic-staging";
+    private static final String STAGING_OWNER = "DroidDeck Epic staging v1\n";
 
     public interface Callback {
         void onProgress(String message, int pct);
@@ -175,6 +187,7 @@ public final class EpicDownloadManager {
             final String cachePath = chunkCacheDirPath == null ? "" : chunkCacheDirPath;
             final File chunkCacheDir = cachePath.isEmpty() ? new File(installDir, ".chunks") : new File(cachePath);
             chunkCacheDir.mkdirs();
+            File stagingDir = recoverOwnedStaging(installDir, manifest.files);
 
             List<FileInfo> selected = resolveInstallFiles(manifest, installTags);
             cb.onLog("epic: " + selected.size() + "/" + manifest.files.size() + " files for tags " + (installTags == null ? "(all)" : installTags.toString()));
@@ -216,99 +229,120 @@ public final class EpicDownloadManager {
                 if (output.isFile()) oldFileBytes.put(f.filename, output.length());
             }
             boolean cacheOnInstallVolume = cachePath.isEmpty();
-            long peakGameAdditional = cacheOnInstallVolume
+            boolean nativeAvailable = StoresNative.INSTANCE.ensureLoaded();
+            long peakGameAdditional = nativeAvailable ? 0 : cacheOnInstallVolume
                     ? estimatePeakAdditionalBytes(pending, chunksByGuid, cachedChunkBytes, true, oldFileBytes)
                     : estimateStagedFilePeak(pending, oldFileBytes);
-            long peakCacheAdditional = cacheOnInstallVolume
+            long peakCacheAdditional = nativeAvailable || cacheOnInstallVolume
                     ? 0 : estimatePeakAdditionalBytes(pending, chunksByGuid, cachedChunkBytes, false);
             long freeGame = StoreInstallRoot.availableBytes(installDir);
-            if (cacheOnInstallVolume) {
+            if (!nativeAvailable && cacheOnInstallVolume) {
                 if (StoreInstallRoot.insufficientSpace(peakGameAdditional, freeGame)) throw new InstallException("Not enough free space for safe staged Epic replacement: need up to " + fmt(peakGameAdditional) + " additional, only " + fmt(freeGame) + " free");
-            } else {
+            } else if (!nativeAvailable) {
                 long freeCache = StoreInstallRoot.availableBytes(chunkCacheDir);
                 if (StoreInstallRoot.insufficientSpace(peakCacheAdditional, freeCache)) throw new InstallException("Not enough internal space for the bounded Epic download cache: need up to " + fmt(peakCacheAdditional) + " additional, only " + fmt(freeCache) + " free");
                 if (StoreInstallRoot.insufficientSpace(peakGameAdditional, freeGame)) throw new InstallException("Not enough free space for safe staged Epic replacement: need " + fmt(peakGameAdditional) + ", only " + fmt(freeGame) + " free");
             }
             if (cancel.get()) return null;
+            if (nativeAvailable) {
+                JSONObject request = nativeRequest(manifest, pending, chunksByGuid, cachedChunkBytes,
+                        oldFileBytes, installDir, chunkCacheDir, stagingDir, cacheOnInstallVolume);
+                request.put("free_game", StoreInstallRoot.availableBytes(installDir));
+                request.put("free_cache", StoreInstallRoot.availableBytes(chunkCacheDir));
+                final long nativeTotal = pendingBytes;
+                final long nativeStarted = System.nanoTime();
+                cb.onSizes(nativeTotal, planned);
+                cb.onBytes(0, nativeTotal, 0);
+                EpicNative.Result outcome = EpicNative.installAdaptive(request.toString(), cancel, new EpicNative.InstallListener() {
+                    @Override public void onPlan(int chunks, long bytes, String mode) {
+                        cb.onLog("epic: engine=adaptive-rust " + mode);
+                    }
+                    @Override public void onProgress(long done, long total, int files, int filesTotal) {
+                        long elapsedMs = Math.max(1, (System.nanoTime() - nativeStarted) / 1_000_000);
+                        cb.onBytes(done, total, done * 1000 / elapsedMs);
+                        cb.onProgress("Downloading chunks…", (int) (done * 80L / Math.max(1, total)));
+                        cb.onStage("stream", done, total, files, filesTotal);
+                    }
+                    @Override public void onLog(String line) { cb.onLog(line); }
+                    @Override public void onComplete(boolean success, String error, long bytes) {}
+                    @Override public boolean beforeSwap() {
+                        try {
+                            if (cancel.get()) return false;
+                            StoreInstalls.markUpdating(installDir);
+                            return !cancel.get();
+                        } catch (Exception e) {
+                            cb.onLog("epic: cannot mark updating: " + e.getMessage());
+                            return false;
+                        }
+                    }
+                });
+                if (outcome.cancelled || cancel.get()) return null;
+                if (!outcome.success) throw new InstallException("Native Epic install failed: " + outcome.error);
+                cb.onStage("install", nativeTotal, nativeTotal, pending.size(), pending.size());
+                cb.onProgress("Complete", 100);
+                return new Result(manifest.launchExe, manifest.buildVersion, planned).withPrereq(manifest);
+            }
             cb.onLog("epic: cache " + (cacheOnInstallVolume ? "beside the game" : "scratch") + ", peak game additional=" + fmt(peakGameAdditional) + (cacheOnInstallVolume ? "" : ", peak cache additional=" + fmt(peakCacheAdditional)) + " for " + fmt(pendingBytes) + " of files");
             final long fTotalBytes = pendingBytes;
             final int totalChunks = needed.size();
             cb.onSizes(fTotalBytes, planned);
             cb.onBytes(0, fTotalBytes, 0);
             cb.onLog("epic: " + fmt(fTotalBytes) + " to write using " + totalChunks + " unique chunks");
-            cb.onLog("epic: engine=bounded-java (shared path; native availability does not change cache semantics)");
-            Map<String, Integer> remainingRefs = chunkReferences(pending);
+            cb.onLog("epic: engine=bounded-java (8-worker prefetch; native availability does not change cache semantics)");
             long completedBytes = 0, writtenBytes = 0;
             long transferStarted = System.nanoTime();
             int completedCount = 0, doneFiles = 0, totalFiles = pending.size();
             Set<String> reservedPaths = new LinkedHashSet<>();
             for (FileInfo file : manifest.files) reservedPaths.add(new File(installDir, file.filename.replace("\\", "/")).getCanonicalPath());
             boolean updateRecorded = false;
-            for (FileInfo file : pending) {
-                if (cancel.get()) return null;
-                String relPath = file.filename.replace("\\", "/");
-                File outFile = new File(installDir, relPath);
-                cb.onProgress("Downloading chunks: " + (relPath.contains("/") ? relPath.substring(relPath.lastIndexOf('/') + 1) : relPath), (int) (completedBytes * 80L / Math.max(1, fTotalBytes)));
-                MessageDigest fileSha = expectedDigest(file.sha1);
-                File partFile = createStagingFile(outFile, reservedPaths);
-                try {
-                    try (FileOutputStream fos = new FileOutputStream(partFile); BufferedOutputStream bos = new BufferedOutputStream(fos, 65536)) {
-                        for (ChunkPart part : file.parts) {
-                            if (cancel.get()) throw new IOException("cancelled");
-                            String guid = part.guidStr();
-                            ChunkInfo chunk = chunksByGuid.get(guid);
-                            if (chunk == null) throw new InstallException("Missing chunk metadata " + guid + " for " + relPath);
-                            File cachedChunk = new File(chunkCacheDir, guid);
-                            if (!cachedChunk.isFile() || !isValidCachedChunk(chunk, cachedChunk)) {
-                                if (cachedChunk.exists() && !cachedChunk.delete() && cachedChunk.exists()) {
-                                    throw new InstallException("Could not clear invalid Epic chunk " + guid);
-                                }
-                                cachedChunkBytes.remove(guid);
-                                cachedChunk = new File(chunkCacheDir, guid);
-                                if (!downloadChunkStreaming(chunk, manifest.chunkDir, cdnUrls, cachedChunk, cancel)) {
-                                    if (cancel.get()) throw new IOException("cancelled");
-                                    throw new InstallException("Failed to download Epic chunk " + guid);
-                                }
-                                cachedChunkBytes.put(guid, cachedChunk.length());
+            try (ChunkPrefetch prefetch = new ChunkPrefetch(pending, chunksByGuid, cachedChunkBytes,
+                    chunkCacheDir, manifest.chunkDir, cdnUrls, cancel)) {
+                for (FileInfo file : pending) {
+                    if (cancel.get()) return null;
+                    String relPath = file.filename.replace("\\", "/");
+                    File outFile = new File(installDir, relPath);
+                    cb.onProgress("Downloading chunks: " + (relPath.contains("/") ? relPath.substring(relPath.lastIndexOf('/') + 1) : relPath), (int) (completedBytes * 80L / Math.max(1, fTotalBytes)));
+                    MessageDigest fileSha = expectedDigest(file.sha1);
+                    File partFile = createStagingFile(outFile, reservedPaths, stagingDir);
+                    try {
+                        try (FileOutputStream fos = new FileOutputStream(partFile); BufferedOutputStream bos = new BufferedOutputStream(fos, 65536)) {
+                            for (ChunkPart part : file.parts) {
+                                if (cancel.get()) throw new IOException("cancelled");
+                                String guid = part.guidStr();
+                                File cachedChunk = prefetch.await(guid);
+                                completedCount++;
+                                cb.onProgress("Downloading chunks (part " + completedCount + ")", (int) (completedBytes * 80L / Math.max(1, fTotalBytes)));
+                                copyChunkPart(cachedChunk, part, bos, fileSha, cancel, relPath);
+                                completedBytes += part.size & 0xFFFFFFFFL;
+                                long elapsedMs = Math.max(1, (System.nanoTime() - transferStarted) / 1_000_000);
+                                cb.onBytes(completedBytes, fTotalBytes, completedBytes * 1000 / elapsedMs);
+                                prefetch.consume();
                             }
-                            completedCount++;
-                            cb.onProgress("Downloading chunks (part " + completedCount + ")", (int) (completedBytes * 80L / Math.max(1, fTotalBytes)));
-                            copyChunkPart(cachedChunk, part, bos, fileSha, cancel, relPath);
-                            completedBytes += part.size & 0xFFFFFFFFL;
-                            long elapsedMs = Math.max(1, (System.nanoTime() - transferStarted) / 1_000_000);
-                            cb.onBytes(completedBytes, fTotalBytes, completedBytes * 1000 / elapsedMs);
-                            if (consumeChunkReference(remainingRefs, guid)) {
-                                if (cachedChunk.exists() && !cachedChunk.delete() && cachedChunk.exists()) {
-                                    throw new InstallException("Could not release consumed Epic chunk " + guid);
-                                }
-                                cachedChunkBytes.remove(guid);
-                            }
-                            trimCacheToBound(chunkCacheDir, cachedChunkBytes);
+                            bos.flush();
+                            fos.getFD().sync();
                         }
-                        bos.flush();
-                        fos.getFD().sync();
+                        if (cancel.get()) return null;
+                        if (partFile.length() != file.fileSize()) throw new InstallException("Size mismatch for " + relPath);
+                        if (fileSha != null && !MessageDigest.isEqual(fileSha.digest(), file.sha1)) {
+                            throw new InstallException("SHA-1 mismatch for " + relPath);
+                        }
+                        // Once a swap can mix versions, the old installed record is no longer truthful.
+                        if (!updateRecorded) {
+                            StoreInstalls.markUpdating(installDir);
+                            updateRecorded = true;
+                        }
+                        if (cancel.get()) return null;
+                        Files.move(partFile.toPath(), outFile.toPath(), StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
+                    } catch (IOException e) {
+                        if (cancel.get()) return null;
+                        throw new InstallException("Could not assemble or atomically replace " + relPath + ": " + e.getMessage());
+                    } finally {
+                        partFile.delete();
                     }
-                    if (cancel.get()) return null;
-                    if (partFile.length() != file.fileSize()) throw new InstallException("Size mismatch for " + relPath);
-                    if (fileSha != null && !MessageDigest.isEqual(fileSha.digest(), file.sha1)) {
-                        throw new InstallException("SHA-1 mismatch for " + relPath);
-                    }
-                    // Once a swap can mix versions, the old installed record is no longer truthful.
-                    if (!updateRecorded) {
-                        StoreInstalls.markUpdating(installDir);
-                        updateRecorded = true;
-                    }
-                    if (cancel.get()) return null;
-                    Files.move(partFile.toPath(), outFile.toPath(), StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
-                } catch (IOException e) {
-                    if (cancel.get()) return null;
-                    throw new InstallException("Could not assemble or atomically replace " + relPath + ": " + e.getMessage());
-                } finally {
-                    partFile.delete();
+                    doneFiles++;
+                    writtenBytes += file.fileSize();
+                    cb.onStage("stream", writtenBytes, pendingBytes, doneFiles, totalFiles);
                 }
-                doneFiles++;
-                writtenBytes += file.fileSize();
-                cb.onStage("stream", writtenBytes, pendingBytes, doneFiles, totalFiles);
             }
             deleteDir(chunkCacheDir);
             cb.onStage("install", writtenBytes, pendingBytes, doneFiles, totalFiles);
@@ -322,12 +356,108 @@ public final class EpicDownloadManager {
         }
     }
 
-    private static File createStagingFile(File output, Set<String> reservedPaths) throws IOException {
+    static File recoverOwnedStaging(File installDir, List<FileInfo> manifestFiles) throws IOException {
+        File staging = new File(installDir, STAGING_DIRECTORY);
+        Path expected = installDir.getCanonicalFile().toPath().resolve(STAGING_DIRECTORY);
+        if (!staging.getCanonicalFile().toPath().equals(expected) ||
+                (Files.exists(staging.toPath(), LinkOption.NOFOLLOW_LINKS) &&
+                        !Files.isDirectory(staging.toPath(), LinkOption.NOFOLLOW_LINKS))) {
+            throw new IOException("Unsafe Epic staging namespace; refusing to traverse it");
+        }
+        String namespace = expected.toString();
+        for (FileInfo file : manifestFiles) {
+            String path = new File(installDir, file.filename.replace("\\", "/")).getCanonicalPath();
+            if (path.equals(namespace) || path.startsWith(namespace + File.separator)) {
+                throw new IOException("Manifest conflicts with Epic staging namespace");
+            }
+        }
+        File owner = new File(staging, "owner");
+        if (Files.exists(staging.toPath(), LinkOption.NOFOLLOW_LINKS)) {
+            if (!Files.isRegularFile(owner.toPath(), LinkOption.NOFOLLOW_LINKS) ||
+                    !owner.getCanonicalFile().toPath().equals(expected.resolve("owner")) ||
+                    owner.length() != STAGING_OWNER.length() ||
+                    !STAGING_OWNER.equals(new String(Files.readAllBytes(owner.toPath()), StandardCharsets.UTF_8))) {
+                throw new IOException("Unowned Epic staging namespace; refusing to delete it");
+            }
+            File[] files = staging.listFiles();
+            if (files == null) throw new IOException("Could not inspect Epic staging");
+            for (File file : files) {
+                if (file.getName().equals("owner")) continue;
+                if (!Files.isRegularFile(file.toPath(), LinkOption.NOFOLLOW_LINKS) ||
+                        !file.getCanonicalFile().toPath().equals(expected.resolve(file.getName())) ||
+                        !file.getName().endsWith(".stage")) throw new IOException("Unexpected Epic staging entry");
+            }
+            for (File file : files) {
+                if (file.getName().equals("owner")) continue;
+                if (!file.delete()) throw new IOException("Could not recover orphan Epic staging");
+            }
+        } else {
+            if (!staging.mkdir()) throw new IOException("Could not create Epic staging namespace");
+            try (FileOutputStream out = new FileOutputStream(owner)) {
+                out.write(STAGING_OWNER.getBytes(StandardCharsets.UTF_8));
+                out.getFD().sync();
+            }
+        }
+        return staging;
+    }
+
+    private static JSONObject nativeRequest(Manifest manifest, List<FileInfo> files,
+            Map<String, ChunkInfo> chunks, Map<String, Long> cached, Map<String, Long> old,
+            File install, File cache, File staging, boolean sameVolume) throws Exception {
+        JSONObject request = new JSONObject();
+        request.put("schema_version", 1);
+        request.put("install_dir", install.getCanonicalPath());
+        request.put("cache_dir", cache.getCanonicalPath());
+        request.put("staging_dir", staging.getCanonicalPath());
+        request.put("same_volume", sameVolume);
+        request.put("chunk_dir", manifest.chunkDir);
+        request.put("max_workers", 32);
+        JSONArray hosts = new JSONArray();
+        for (CdnUrl cdn : manifest.cdnUrls) hosts.put(cdn.baseUrl + cdn.cloudDir);
+        request.put("hosts", hosts);
+        JSONArray chunkList = new JSONArray();
+        for (ChunkInfo chunk : chunks.values()) {
+            JSONObject item = new JSONObject();
+            item.put("guid", chunk.guidStr());
+            item.put("path", chunk.getPath(manifest.chunkDir));
+            item.put("size", chunkWindowSize(chunk));
+            item.put("exact_size", chunk.windowSize > 0);
+            item.put("wire_size", Math.max(1, chunk.fileSize));
+            item.put("sha1", hexDigest(chunk.sha1));
+            item.put("cached", cached.containsKey(chunk.guidStr()) ? cached.get(chunk.guidStr()) : -1);
+            chunkList.put(item);
+        }
+        request.put("chunks", chunkList);
+        JSONArray fileList = new JSONArray();
+        for (FileInfo file : files) {
+            JSONObject item = new JSONObject();
+            item.put("path", file.filename.replace("\\", "/"));
+            item.put("sha1", hexDigest(file.sha1));
+            item.put("old_size", old.containsKey(file.filename) ? old.get(file.filename) : 0);
+            JSONArray parts = new JSONArray();
+            for (ChunkPart part : file.parts) {
+                parts.put(new JSONObject().put("guid", part.guidStr()).put("offset", part.offset & 0xFFFFFFFFL).put("size", part.size & 0xFFFFFFFFL));
+            }
+            item.put("parts", parts);
+            fileList.put(item);
+        }
+        request.put("files", fileList);
+        return request;
+    }
+
+    private static String hexDigest(byte[] digest) {
+        if (digest == null || digest.length != 20) return "";
+        StringBuilder out = new StringBuilder();
+        for (byte b : digest) out.append(String.format("%02x", b & 255));
+        return out.toString();
+    }
+
+    private static File createStagingFile(File output, Set<String> reservedPaths, File staging) throws IOException {
         File parent = output.getParentFile();
         if (!parent.isDirectory() && !parent.mkdirs()) throw new IOException("Could not create output directory");
-        // Unique sibling staging is on the output volume, never a manifest's literal "<name>.part".
+        // Owned staging is on the output volume, never a manifest's literal "<name>.part".
         while (true) {
-            File stage = File.createTempFile(".droiddeck-epic-", ".stage", parent);
+            File stage = File.createTempFile("file-", ".stage", staging);
             if (!reservedPaths.contains(stage.getCanonicalPath())) return stage;
             if (!stage.delete()) throw new IOException("Could not clear staging collision");
         }
@@ -407,6 +537,135 @@ public final class EpicDownloadManager {
         return false;
     }
 
+    /**
+     * Pure disk schedule shared by execution and preflight. The next eight part references pin
+     * at most eight GUIDs (including across file boundaries); other reusable data is oldest-first
+     * capped at 64 MiB. Incoming downloads reserve their entire decoded size before submission.
+     */
+    static final class PrefetchSchedule {
+        final List<String> parts = new ArrayList<>();
+        final Map<String, Long> resident;
+        final Map<String, Integer> refs;
+        final Map<String, ChunkInfo> chunks;
+        int cursor;
+
+        PrefetchSchedule(List<FileInfo> files, Map<String, ChunkInfo> chunks, Map<String, Long> cached) {
+            for (FileInfo file : files) for (ChunkPart part : file.parts) parts.add(part.guidStr());
+            this.chunks = chunks;
+            resident = new LinkedHashMap<>(cached);
+            refs = chunkReferences(files);
+        }
+
+        Set<String> window() {
+            return new LinkedHashSet<>(parts.subList(cursor, Math.min(parts.size(), cursor + PREFETCH_PARTS)));
+        }
+
+        List<String> fill() {
+            List<String> fetch = new ArrayList<>();
+            for (String guid : window()) if (!resident.containsKey(guid)) {
+                ChunkInfo chunk = chunks.get(guid);
+                if (chunk == null) throw new IllegalArgumentException("Missing chunk metadata " + guid);
+                resident.put(guid, chunkWindowSize(chunk));
+                fetch.add(guid);
+            }
+            return fetch;
+        }
+
+        List<String> consume() {
+            List<String> release = new ArrayList<>();
+            String guid = parts.get(cursor++);
+            if (consumeChunkReference(refs, guid)) {
+                resident.remove(guid);
+                release.add(guid);
+            }
+            Set<String> pinned = window();
+            long reusable = 0;
+            for (Map.Entry<String, Long> entry : resident.entrySet()) {
+                if (!pinned.contains(entry.getKey())) reusable += entry.getValue();
+            }
+            Iterator<Map.Entry<String, Long>> it = resident.entrySet().iterator();
+            while (reusable > MAX_RETAINED_CHUNK_CACHE_BYTES && it.hasNext()) {
+                Map.Entry<String, Long> entry = it.next();
+                if (pinned.contains(entry.getKey())) continue;
+                reusable -= entry.getValue();
+                release.add(entry.getKey());
+                it.remove();
+            }
+            return release;
+        }
+
+        long bytes() {
+            long bytes = 0;
+            for (long size : resident.values()) bytes += size;
+            return bytes;
+        }
+    }
+
+    private static final class ChunkPrefetch implements AutoCloseable {
+        final PrefetchSchedule schedule;
+        final File cacheDir;
+        final AtomicBoolean cancel;
+        final AtomicBoolean stopped = new AtomicBoolean();
+        final ExecutorService workers = Executors.newFixedThreadPool(PREFETCH_PARTS);
+        final Map<String, Future<Boolean>> jobs = new LinkedHashMap<>();
+        final Set<HttpURLConnection> connections = ConcurrentHashMap.newKeySet();
+        final String chunkDir;
+        final List<CdnUrl> cdns;
+
+        ChunkPrefetch(List<FileInfo> files, Map<String, ChunkInfo> chunks, Map<String, Long> cached,
+                      File cacheDir, String chunkDir, List<CdnUrl> cdns, AtomicBoolean cancel) {
+            schedule = new PrefetchSchedule(files, chunks, cached);
+            this.cacheDir = cacheDir;
+            this.chunkDir = chunkDir;
+            this.cdns = cdns;
+            this.cancel = cancel;
+        }
+
+        File await(String guid) throws Exception {
+            for (String fetch : schedule.fill()) {
+                jobs.put(fetch, workers.submit(() -> downloadChunkStreaming(schedule.chunks.get(fetch),
+                        chunkDir, cdns, new File(cacheDir, fetch), cancel, stopped, connections)));
+            }
+            Future<Boolean> job = jobs.get(guid);
+            if (job != null) while (true) {
+                if (cancel.get()) throw new IOException("cancelled");
+                try {
+                    if (!job.get(100, TimeUnit.MILLISECONDS)) {
+                        if (cancel.get()) throw new IOException("cancelled");
+                        throw new InstallException("Failed to download Epic chunk " + guid);
+                    }
+                    break;
+                } catch (TimeoutException ignored) { }
+            }
+            return new File(cacheDir, guid);
+        }
+
+        void consume() throws InstallException {
+            for (String guid : schedule.consume()) {
+                // Only the writer advances the schedule, after closing its chunk reader.
+                jobs.remove(guid);
+                File chunk = new File(cacheDir, guid);
+                if (chunk.exists() && !chunk.delete() && chunk.exists()) {
+                    throw new InstallException("Could not release Epic cache chunk " + guid);
+                }
+            }
+        }
+
+        @Override public void close() {
+            stopped.set(true);
+            for (HttpURLConnection connection : connections) connection.disconnect();
+            workers.shutdownNow();
+            boolean interrupted = false;
+            while (true) {
+                try {
+                    if (workers.awaitTermination(100, TimeUnit.MILLISECONDS)) break;
+                    for (HttpURLConnection connection : connections) connection.disconnect();
+                } catch (InterruptedException e) { interrupted = true; }
+            }
+            if (interrupted) Thread.currentThread().interrupt();
+        }
+    }
+
     static boolean isValidCachedChunk(ChunkInfo chunk, File file) {
         if (!file.isFile()) return false;
         if (chunk.windowSize > 0 && file.length() != chunk.windowSize) return false;
@@ -445,7 +704,7 @@ public final class EpicDownloadManager {
     }
 
     /**
-     * Simulates the serial fetch → append part → last-reference eviction order. The returned
+     * Simulates reserved parallel prefetch → append part → last-reference eviction order. The returned
      * additional-space peak is measured against the current free-space reading: existing verified
      * chunks are already reflected in that reading and count as reclaimed only when actually freed.
      * Old output bytes are reclaimed after the complete new file is staged and verified, never
@@ -459,33 +718,21 @@ public final class EpicDownloadManager {
     static long estimatePeakAdditionalBytes(List<FileInfo> files, Map<String, ChunkInfo> chunksByGuid,
                                             Map<String, Long> cachedChunkBytes, boolean includeFileWrites,
                                             Map<String, Long> oldFileBytes) {
-        Map<String, Integer> refs = chunkReferences(files);
-        Map<String, Long> residentChunks = new LinkedHashMap<>(cachedChunkBytes);
-        long residentDelta = 0, peak = 0;
+        PrefetchSchedule schedule = new PrefetchSchedule(files, chunksByGuid, cachedChunkBytes);
+        long initialCache = schedule.bytes();
+        long fileDelta = 0, peak = 0;
         for (FileInfo file : files) {
             for (ChunkPart part : file.parts) {
-                String guid = part.guidStr();
-                ChunkInfo chunk = chunksByGuid.get(guid);
-                if (chunk == null) throw new IllegalArgumentException("Missing chunk metadata " + guid);
-                Long knownBytes = residentChunks.get(guid);
-                long chunkBytes = knownBytes != null ? knownBytes : chunkWindowSize(chunk);
-                if (knownBytes == null) {
-                    residentChunks.put(guid, chunkBytes);
-                    residentDelta += chunkBytes;
-                    peak = Math.max(peak, residentDelta);
-                }
+                schedule.fill();
+                peak = Math.max(peak, fileDelta + schedule.bytes() - initialCache);
                 if (includeFileWrites) {
-                    residentDelta += part.size & 0xFFFFFFFFL;
-                    peak = Math.max(peak, residentDelta);
+                    fileDelta += part.size & 0xFFFFFFFFL;
+                    peak = Math.max(peak, fileDelta + schedule.bytes() - initialCache);
                 }
-                if (consumeChunkReference(refs, guid)) {
-                    Long removed = residentChunks.remove(guid);
-                    if (removed != null) residentDelta -= removed;
-                }
-                for (long bytes : evictOldestReusableChunks(residentChunks).values()) residentDelta -= bytes;
+                schedule.consume();
             }
             // The verified staging file coexists with the old output until its atomic swap.
-            if (includeFileWrites) residentDelta -= oldFileBytes.containsKey(file.filename) ? oldFileBytes.get(file.filename) : 0;
+            if (includeFileWrites) fileDelta -= oldFileBytes.containsKey(file.filename) ? oldFileBytes.get(file.filename) : 0;
         }
         return Math.max(0, peak);
     }
@@ -786,15 +1033,19 @@ public final class EpicDownloadManager {
     }
 
     /** One chunk streamed from the first CDN that serves it, SHA-1 checked, published by rename. */
-    private static boolean downloadChunkStreaming(ChunkInfo chunk, String chunkDir, List<CdnUrl> cdnUrls, File outFile, AtomicBoolean cancel) {
+    private static boolean downloadChunkStreaming(ChunkInfo chunk, String chunkDir, List<CdnUrl> cdnUrls,
+                                                  File outFile, AtomicBoolean cancel, AtomicBoolean stopped,
+                                                  Set<HttpURLConnection> connections) {
         String chunkPath = chunk.getPath(chunkDir);
         File tmp = new File(outFile.getPath() + ".part");
         tmp.delete();
         for (CdnUrl cdn : cdnUrls) {
-            if (cancel.get()) return false;
+            if (cancel.get() || stopped.get()) return false;
             HttpURLConnection conn = null;
             try {
                 conn = (HttpURLConnection) new URL(cdn.baseUrl + cdn.cloudDir + "/" + chunkPath).openConnection();
+                connections.add(conn);
+                if (cancel.get() || stopped.get()) return false;
                 conn.setConnectTimeout(30000);
                 conn.setReadTimeout(60000);
                 conn.setRequestProperty("User-Agent", UA);
@@ -824,7 +1075,7 @@ public final class EpicDownloadManager {
                         int remaining = compressedSize;
                         try {
                             while (remaining > 0 && !inflater.finished()) {
-                                if (cancel.get()) break;
+                                if (cancel.get() || stopped.get()) break;
                                 if (inflater.needsInput()) {
                                     int n = in.read(iobuf, 0, Math.min(iobuf.length, remaining));
                                     if (n <= 0) break;
@@ -843,18 +1094,19 @@ public final class EpicDownloadManager {
                             }
                             int out;
                             while ((out = inflater.inflate(obuf)) > 0) {
+                                if (cancel.get() || stopped.get()) break;
                                 outputBytes = checkedChunkWriteSize(chunk, outputBytes, out);
                                 fos.write(obuf, 0, out);
                                 if (sha != null) sha.update(obuf, 0, out);
                             }
-                            if (!cancel.get() && !inflater.finished()) throw new IOException("truncated compressed Epic chunk");
+                            if (!cancel.get() && !stopped.get() && !inflater.finished()) throw new IOException("truncated compressed Epic chunk");
                         } finally {
                             inflater.end();
                         }
                     } else {
                         int remaining = compressedSize;
                         while (remaining > 0) {
-                            if (cancel.get()) break;
+                            if (cancel.get() || stopped.get()) break;
                             int n = in.read(iobuf, 0, Math.min(iobuf.length, remaining));
                             if (n <= 0) break;
                             outputBytes = checkedChunkWriteSize(chunk, outputBytes, n);
@@ -862,10 +1114,10 @@ public final class EpicDownloadManager {
                             if (sha != null) sha.update(iobuf, 0, n);
                             remaining -= n;
                         }
-                        if (!cancel.get() && remaining != 0) throw new IOException("truncated Epic chunk");
+                        if (!cancel.get() && !stopped.get() && remaining != 0) throw new IOException("truncated Epic chunk");
                     }
                 }
-                if (cancel.get()) { tmp.delete(); conn.disconnect(); return false; }
+                if (cancel.get() || stopped.get()) { tmp.delete(); conn.disconnect(); return false; }
                 if (chunk.windowSize > 0 && tmp.length() != chunk.windowSize) { tmp.delete(); conn.disconnect(); continue; }
                 if (sha != null && !MessageDigest.isEqual(sha.digest(), chunk.sha1)) { tmp.delete(); conn.disconnect(); continue; }
                 if (!tmp.renameTo(outFile)) { tmp.delete(); conn.disconnect(); continue; }
@@ -875,6 +1127,12 @@ public final class EpicDownloadManager {
                 Log.w(TAG, "chunk " + chunk.guidStr() + " failed: " + e.getMessage());
                 tmp.delete();
                 if (conn != null) conn.disconnect();
+            } finally {
+                if (conn != null) {
+                    connections.remove(conn);
+                    conn.disconnect();
+                }
+                tmp.delete();
             }
         }
         tmp.delete();

app/src/main/java/com/droiddeck/launcher/stores/epic/EpicNative.kt

@@ -9,18 +9,17 @@ import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 
 /**
- * The native batch APIs for Epic in `libdroiddeckstores.so`: [run] fills the chunk cache with
- * verified, decompressed chunks for the files the manager says are pending, [assemble] writes those
- * files from the cache. The production installer intentionally uses one serial Java
- * chunk-to-file path instead, so cache bounds and resume behavior do not depend on native
- * availability. These APIs remain for callers that explicitly need batch fetch/assembly.
+ * [installAdaptive] is the production native Epic fetch/assembly engine. Its snapshot is the
+ * manager's delta-selected pending files, verified cache and old output sizes. Rust chooses a
+ * storage-safe ordered lookahead, overlaps fetch_core with staged assembly and calls beforeSwap
+ * before the first verified replacement. A loaded-engine error fails explicitly.
  *
  * The cache is `<installDir>/.chunks` when `chunkCacheDir` is `""`, else the scratch folder given -
- * the app's own cache when the game installs to a card, where the assembler drops each chunk right
- * after its last use and the folder at the end. Both calls take the same value for one install.
+ * the app's own cache when the game installs to a card. Adaptive installs release last-use chunks
+ * in both layouts and remove the cache on success. [run]/[assemble] are legacy batch APIs.
  *
- * Both block while polling [AtomicBoolean] `cancel` every 250 ms: on cancel they flip the native
- * flag, wait up to 5 s for the run to wind down, then return.
+ * Cancellation flips the native flag; the native handle is joined before returning, so no
+ * cache/staging writer survives completion or cache cleanup.
  */
 object EpicNative {
     private const val TAG = "EpicNative"
@@ -37,6 +36,18 @@ object EpicNative {
         fun onComplete(success: Boolean, error: String, bytes: Long)
     }
 
+    interface InstallListener : Listener {
+        /** Synchronous approval before the first verified atomic replacement. False aborts. */
+        fun beforeSwap(): Boolean
+    }
+
+    /** One native budgeted fetch/assembly run. A loaded-engine failure is never a Java retry. */
+    @JvmStatic
+    fun installAdaptive(request: String, cancel: AtomicBoolean, listener: InstallListener): Result =
+        drive("nativeInstallAdaptive", cancel, listener, strict = true) { inner ->
+            nativeInstallAdaptive(request, inner)
+        }
+
     /** Outcome of a run. `started == false`: nothing happened on disk, the caller runs its own loop. */
     class Result(
         @JvmField val started: Boolean, @JvmField val success: Boolean, @JvmField val cancelled: Boolean,
@@ -59,13 +70,14 @@ object EpicNative {
     fun assemble(manifest: ByteArray, installDir: String, chunkCacheDir: String, fileIdx: IntArray, cancel: AtomicBoolean?, listener: Listener): Result =
         drive("nativeAssemble", cancel, listener) { inner -> nativeAssemble(manifest, installDir, chunkCacheDir, fileIdx, inner) }
 
-    private fun drive(what: String, cancel: AtomicBoolean?, listener: Listener, start: (Listener) -> Long): Result {
+    private fun drive(what: String, cancel: AtomicBoolean?, listener: Listener, strict: Boolean = false, start: (Listener) -> Long): Result {
         if (!StoresNative.ensureLoaded()) return Result(false, false, false, "engine not built", 0L, 0, 0)
         val latch = CountDownLatch(1)
         val completion = AtomicReference<Completion?>(null)
         val doneRef = AtomicReference(0)
         val totalRef = AtomicReference(0)
-        val inner = object : Listener {
+        val inner = object : InstallListener {
+            override fun beforeSwap(): Boolean = (listener as? InstallListener)?.beforeSwap() ?: false
             override fun onPlan(chunksTotal: Int, bytesTotal: Long, chunkDir: String) { totalRef.set(chunksTotal); listener.onPlan(chunksTotal, bytesTotal, chunkDir) }
             override fun onProgress(bytesDone: Long, bytesTotal: Long, done: Int, total: Int) { doneRef.set(done); totalRef.set(total); listener.onProgress(bytesDone, bytesTotal, done, total) }
             override fun onLog(line: String) = listener.onLog(line)
@@ -78,21 +90,26 @@ object EpicNative {
         val handle: Long = try {
             start(inner)
         } catch (t: Throwable) {
-            Log.w(TAG, "engine unavailable: ${t.javaClass.simpleName}: ${t.message}")
-            return Result(false, false, false, "$what: ${t.javaClass.simpleName}", 0L, 0, 0)
+            Log.w(TAG, "${if (strict) "engine start failed (no fallback)" else "engine unavailable"}: ${t.javaClass.simpleName}: ${t.message}")
+            return Result(strict, false, false, "$what: ${t.javaClass.simpleName}: ${t.message}", 0L, 0, 0)
         }
-        if (handle == 0L) return Result(false, false, false, completion.get()?.error ?: "not started", 0L, 0, totalRef.get())
+        if (handle == 0L) return Result(strict, false, false, completion.get()?.error ?: "not started", 0L, 0, totalRef.get())
         var cancelSent = false
         try {
             while (!latch.await(250, TimeUnit.MILLISECONDS)) {
                 if (cancel != null && cancel.get()) {
                     nativeCancel(handle)
                     cancelSent = true
-                    latch.await(5, TimeUnit.SECONDS)
+                    if (strict) {
+                        while (!latch.await(250, TimeUnit.MILLISECONDS)) { }
+                    } else {
+                        latch.await(5, TimeUnit.SECONDS)
+                    }
                     break
                 }
             }
         } finally {
+            if (completion.get() == null) nativeCancel(handle)
             nativeRelease(handle)
         }
         val c = completion.get()
@@ -101,6 +118,9 @@ object EpicNative {
         return Result(true, success, cancelSent, error, c?.bytes ?: 0L, doneRef.get(), totalRef.get())
     }
 
+    @JvmStatic
+    private external fun nativeInstallAdaptive(request: String, listener: Listener): Long
+
     @JvmStatic
     private external fun nativeStart(
         manifest: ByteArray, installDir: String, chunkCacheDir: String, cdnPrefixes: Array<String>, pendingFileIdx: IntArray, expectedChunks: Int, expectedBytes: Long,

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

@@ -112,7 +112,37 @@ 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:
+### Production adaptive install (snapshot schema v1)
+
+`nativeInstallAdaptive(request: String, listener: EpicNative.Listener): Long`
+exports `Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeInstallAdaptive`.
+Its exact Kotlin descriptor is
+`(Ljava/lang/String;Lcom/droiddeck/launcher/stores/epic/EpicNative$Listener;)J`.
+The listener must also implement `EpicNative.InstallListener.beforeSwap(): Boolean` (`()Z`).
+Rust invokes it synchronously after staging/hash/sync and before the first atomic replacement;
+false or a Java exception aborts without replacing that output. Java marks the sidecar updating
+there. Later verified swaps preserve mixed-update/resume semantics.
+
+The UTF-8 JSON request contains:
+
+- `schema_version: 1`, absolute `install_dir`, `cache_dir`, owned `staging_dir`,
+  `same_volume`, signed `free_game`/`free_cache` (`-1` unknown), `max_workers`, CDN `hosts`.
+- `chunks`: distinct `guid`, relative CDN `path`, decoded `size`, `exact_size`, `wire_size`,
+  SHA-1 hex `sha1` (empty/all-zero means absent), `cached` verified length (`-1` absent).
+- `files`: pending manifest-order relative `path`, `sha1`, current `old_size`, ordered `parts`
+  containing `guid`, unsigned `offset` and `size`.
+
+Parse, owned-namespace validation and adaptive storage planning happen synchronously. Errors
+return zero after `onComplete(false, error, 0)` and are **terminal**, not fallback signals.
+`onPlan` reports the selected lookahead/retention/peaks; `onProgress` counts staged file bytes
+and verified swapped file counts, not fetched wire bytes. Fetch and staged assembly overlap
+in one native run, using the same pure reservation/release policy as the peak planner.
+Cancellation/release joins native workers; no worker survives the terminal callback/handle
+release. Only `StoresNative.ensureLoaded() == false` selects bounded parallel Java.
+
+### Legacy batch APIs (not the production installer)
+
+Two explicit calls provide the legacy batch interface:
 
 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

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

@@ -1,15 +1,14 @@
 //! Epic Games Store download adapter for the shared fetch core (`crate::fetch_core`).
 //!
-//! Scope = the chunk-fetch inner loop of `EpicDownloadManager.java` (the fixed 8-thread pool that
-//! fills `<installDir>/.chunks/<GUID>`), nothing more. The Java manager still parses the manifest
-//! API JSON, downloads the manifest, selects files (install tags), runs the delta/verify pass,
-//! assembles files from the chunk cache and does every post-install step. The adapter re-parses
-//! the same manifest bytes, rebuilds the same chunk plan for the pending files Java hands it,
-//! skips chunks already in the cache and writes verified chunks with the same `.part` + rename
-//! protocol, so the on-disk state is identical whichever engine ran (see
-//! Bannerlator's `docs/RUST_EPIC_PARITY.md` for the rule-by-rule table).
+//! Production installs use [`adaptive`]: the Java manifest/delta pass supplies a versioned
+//! snapshot, the shared pure storage schedule selects an ordered lookahead, and fetch_core
+//! runs concurrently with verified staged assembly. Old outputs survive until atomic swap;
+//! a synchronous JNI callback marks the sidecar updating before the first replacement.
+//! Legacy manifest/batch APIs remain available. A missing library uses the bounded Java
+//! installer; native plan/runtime failures never silently switch engines.
 //!
 //! Submodules:
+//! - [`adaptive`] — budget planner and bounded overlapping fetch/assembly coordinator.
 //! - [`manifest`] — ChunksV4 binary manifest + legacy JSON manifest parsers (1:1 with the Java
 //!   `parseManifest` / `parseJsonManifest`).
 //! - [`plan`] — install-tag selection, unique-chunk plan, chunk paths / CDN URLs.
@@ -26,6 +25,7 @@
 //! carries the same licence (see `Cargo.toml`).
 
 pub mod assemble;
+pub mod adaptive;
 pub mod chunk;
 pub mod driver;
 pub mod jni;

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

@@ -24,7 +24,7 @@
 
 use std::collections::HashMap;
 use std::fs::{self, File};
-use std::io::{BufWriter, Write};
+use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 
@@ -33,6 +33,33 @@ use sha1::{Digest, Sha1};
 use super::manifest::{parse_manifest, FileInfo, Manifest};
 use super::plan::resolve_cache_dir;
 
+/// Streaming assembly primitive used by the budgeted installer; never holds a decoded chunk
+/// in memory and closes the reader before the caller releases the cache reservation.
+pub fn copy_chunk_part(
+    cache_path: &Path, offset: u64, size: u64, output: &mut impl Write,
+    sha: &mut Sha1, cancel: &AtomicBool,
+) -> Result<u64, String> {
+    let mut input = File::open(cache_path).map_err(|e| format!("read chunk: {e}"))?;
+    let length = input.metadata().map_err(|e| e.to_string())?.len();
+    if offset.checked_add(size).filter(|end| *end <= length).is_none() {
+        return Err("chunk part exceeds cached data".into());
+    }
+    input.seek(SeekFrom::Start(offset)).map_err(|e| e.to_string())?;
+    let mut left = size;
+    let mut buffer = [0u8; 64 * 1024];
+    while left > 0 {
+        if cancel.load(Ordering::Acquire) { return Err("cancelled".into()); }
+        let take = left.min(buffer.len() as u64) as usize;
+        let count = input.read(&mut buffer[..take])
+            .map_err(|e| e.to_string())?;
+        if count == 0 { return Err("short cached chunk".into()); }
+        output.write_all(&buffer[..count]).map_err(|e| e.to_string())?;
+        sha.update(&buffer[..count]);
+        left -= count as u64;
+    }
+    Ok(size)
+}
+
 /// Everything the JNI layer hands over for one assembly run.
 pub struct AssembleRequest {
     pub manifest_bytes: Vec<u8>,

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

@@ -93,10 +93,14 @@ struct HashingWriter<W: Write> {
     inner: W,
     sha: Option<Sha1>,
     written: u64,
+    limit: u64,
 }
 
 impl<W: Write> Write for HashingWriter<W> {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        if self.written.saturating_add(buf.len() as u64) > self.limit {
+            return Err(io::Error::other("chunk exceeds reserved decoded window"));
+        }
         let n = self.inner.write(buf)?;
         if let Some(sha) = self.sha.as_mut() {
             sha.update(&buf[..n]);
@@ -129,10 +133,21 @@ pub fn write_verified_chunk(
     // `tmp.delete()` — clear any stale partial from a prior interrupted attempt.
     let _ = fs::remove_file(&tmp);
 
-    let result = write_part(body, expected_sha1, &tmp, final_path);
+    let result = write_part(body, expected_sha1, &tmp, final_path, u64::MAX, false);
     if result.is_err() {
         let _ = fs::remove_file(&tmp);
     }
+
+    result
+}
+
+pub fn write_verified_chunk_bounded(
+    body: &[u8], expected_sha1: Option<&[u8; 20]>, final_path: &Path,
+    decoded_limit: u64, exact_size: bool,
+) -> Result<u64, String> {
+    let tmp = part_path(final_path);
+    let result = write_part(body, expected_sha1, &tmp, final_path, decoded_limit, exact_size);
+    if result.is_err() { let _ = fs::remove_file(tmp); }
     result
 }
 
@@ -141,23 +156,46 @@ fn write_part(
     expected_sha1: Option<&[u8; 20]>,
     tmp: &Path,
     final_path: &Path,
+    limit: u64,
+    exact_size: bool,
 ) -> Result<u64, String> {
     let hdr = parse_chunk_header(body)?;
     let payload = chunk_payload(body, &hdr)?;
+    if limit != u64::MAX && (hdr.compressed_size < 0 || payload.len() != hdr.compressed_size as usize) {
+        return Err("truncated chunk body".into());
+    }
 
     let file = File::create(tmp).map_err(|e| format!("open {}: {e}", tmp.display()))?;
     let mut out = HashingWriter {
         inner: BufWriter::with_capacity(128 * 1024, file),
         sha: expected_sha1.map(|_| Sha1::new()),
         written: 0,
+        limit,
     };
 
     if hdr.is_compressed() {
         // Java: inflate until the stream finishes or the input runs out; a corrupt stream throws
         // (DataFormatException) → this CDN attempt fails. flate2 behaves the same way: EOF on a
         // truncated stream is a clean end (partial output), corruption is an error.
-        let mut dec = flate2::read::ZlibDecoder::new(payload);
-        io::copy(&mut dec, &mut out).map_err(|e| format!("inflate: {e}"))?;
+        if limit == u64::MAX {
+            let mut dec = flate2::read::ZlibDecoder::new(payload);
+            io::copy(&mut dec, &mut out).map_err(|e| format!("inflate: {e}"))?;
+        } else {
+            let mut dec = flate2::Decompress::new(true);
+            let mut buffer = [0u8; 128 * 1024];
+            loop {
+                let input = dec.total_in() as usize;
+                let before = dec.total_out();
+                let status = dec.decompress(&payload[input..], &mut buffer, flate2::FlushDecompress::None)
+                    .map_err(|e| format!("inflate: {e}"))?;
+                let count = (dec.total_out() - before) as usize;
+                out.write_all(&buffer[..count]).map_err(|e| format!("inflate write: {e}"))?;
+                if status == flate2::Status::StreamEnd { break; }
+                if count == 0 && dec.total_in() as usize == input {
+                    return Err("truncated compressed Epic chunk".into());
+                }
+            }
+        }
     } else {
         out.write_all(payload).map_err(|e| format!("write: {e}"))?;
     }
@@ -165,6 +203,9 @@ fn write_part(
     let written = out.written;
     let digest = out.sha.take().map(|s| s.finalize());
     drop(out); // close the file before the rename
+    if exact_size && written != limit {
+        return Err("chunk decoded size mismatch".into());
+    }
 
     if let (Some(expected), Some(actual)) = (expected_sha1, digest) {
         if actual.as_slice() != &expected[..] {

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

@@ -63,6 +63,7 @@ fn android_log(message: &str) {
 /// assembly run, so one `nativeCancel` / `nativeRelease` pair serves both.
 pub struct EpicDownloadHandle {
     cancel: Arc<AtomicBool>,
+    worker: Option<thread::JoinHandle<()>>,
 }
 
 fn clear_pending_exception(env: &mut JNIEnv) {
@@ -314,12 +315,11 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
 
     let cancel = Arc::new(AtomicBool::new(false));
     let thread_cancel = Arc::clone(&cancel);
-    let handle = Box::new(EpicDownloadHandle { cancel });
-
-    thread::spawn(move || {
+    let worker = thread::spawn(move || {
         crate::priority::background();
         run_on_thread(vm, listener, plan, req, thread_cancel);
     });
+    let handle = Box::new(EpicDownloadHandle { cancel, worker: Some(worker) });
 
     Box::into_raw(handle) as jlong
 }
@@ -425,9 +425,7 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
 
     let cancel = Arc::new(AtomicBool::new(false));
     let thread_cancel = Arc::clone(&cancel);
-    let handle = Box::new(EpicDownloadHandle { cancel });
-
-    thread::spawn(move || {
+    let worker = thread::spawn(move || {
         crate::priority::background();
         let progress = |bytes_done: u64, bytes_total: u64, files_done: u64, files_total: u64| {
             let Ok(mut env) = vm.attach_current_thread_as_daemon() else {
@@ -454,6 +452,7 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
             outcome.bytes_written,
         );
     });
+    let handle = Box::new(EpicDownloadHandle { cancel, worker: Some(worker) });
 
     Box::into_raw(handle) as jlong
 }
@@ -483,5 +482,73 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
     }
     // The worker thread owns its own Arc clone of the cancel flag and its own GlobalRef, so
     // freeing the handle is safe even if a (cancelled) run is still winding down.
-    drop(unsafe { Box::from_raw(handle as *mut EpicDownloadHandle) });
+    let mut handle = unsafe { Box::from_raw(handle as *mut EpicDownloadHandle) };
+    if let Some(worker) = handle.worker.take() { let _ = worker.join(); }
+}
+
+/// ABI v1 snapshot: selected chunks/files, verified cache lengths, old output lengths and both
+/// measured free-space probes. Parse/budget failures are terminal, never fallback requests.
+#[no_mangle]
+pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeInstallAdaptive(
+    mut env: JNIEnv, _class: JClass, request: JString, listener: JObject,
+) -> jlong {
+    if listener.is_null() { return 0; }
+    let parsed = jstring_to_string(&mut env, &request).ok_or("unreadable adaptive request".to_string())
+        .and_then(|s| super::adaptive::Request::parse(&s))
+        .and_then(|req| super::adaptive::choose_budget(&req).map(|budget| (req, budget)));
+    let (req, budget) = match parsed {
+        Ok(plan) => plan,
+        Err(error) => {
+            call_on_complete(&mut env, &listener, false, &error, 0);
+            return 0;
+        }
+    };
+    call_on_plan(&mut env, &listener, req.chunks.len() as u64, 0,
+        &format!("lookahead={} retained={} game_peak={} cache_peak={}",
+            budget.lookahead, budget.retained, budget.game_peak, budget.cache_peak));
+    let Ok(vm) = env.get_java_vm() else {
+        call_on_complete(&mut env, &listener, false, "JavaVM unavailable", 0);
+        return 0;
+    };
+    let Ok(listener) = env.new_global_ref(&listener) else {
+        call_on_complete(&mut env, &listener, false, "listener ref failed", 0);
+        return 0;
+    };
+    let cancel = Arc::new(AtomicBool::new(false));
+    let thread_cancel = cancel.clone();
+    let worker = thread::spawn(move || {
+        crate::priority::background();
+        let progress = |done, total, files, files_total| {
+            if let Ok(mut env) = vm.attach_current_thread_as_daemon() {
+                call_on_progress(&mut env, listener.as_obj(), done, total, files, files_total);
+            }
+        };
+        let before_swap = || -> Result<(), String> {
+            let mut env = vm.attach_current_thread_as_daemon().map_err(|e| e.to_string())?;
+            match env.call_method(listener.as_obj(), "beforeSwap", "()Z", &[]) {
+                Ok(value) => if value.z().unwrap_or(false) { Ok(()) } else { Err("Could not mark installed record updating before replacement".into()) },
+                _ => {
+                    clear_pending_exception(&mut env);
+                    Err("Could not mark installed record updating before replacement".into())
+                }
+            }
+        };
+        let log = |line: &str| {
+            android_log(line);
+            if let Ok(mut env) = vm.attach_current_thread_as_daemon() {
+                call_on_log(&mut env, listener.as_obj(), line);
+            }
+        };
+        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(||
+            super::adaptive::run(&req, budget, &thread_cancel, &before_swap, &progress, &log)))
+            .unwrap_or_else(|_| Err("Native Epic install panicked".into()));
+        let (success, error, bytes) = match outcome {
+            Ok(bytes) => (true, String::new(), bytes),
+            Err(error) => (false, error, 0),
+        };
+        if let Ok(mut env) = vm.attach_current_thread_as_daemon() {
+            call_on_complete(&mut env, listener.as_obj(), success, &error, bytes);
+        }
+    });
+    Box::into_raw(Box::new(EpicDownloadHandle { cancel, worker: Some(worker) })) as jlong
 }

app/src/test/java/com/droiddeck/launcher/stores/epic/EpicBoundedInstallTest.kt

@@ -12,6 +12,11 @@ import java.nio.ByteOrder
 import java.security.MessageDigest
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.net.Socket
 import org.junit.After
 import org.junit.Assert.*
 import org.junit.Before
@@ -33,6 +38,11 @@ class EpicBoundedInstallTest {
     private lateinit var root: File
     private val requests = AtomicInteger()
     private val reject = AtomicBoolean()
+    private val rejectGuid = AtomicInteger()
+    private val handlers = Executors.newCachedThreadPool()
+    private val sockets = ConcurrentHashMap.newKeySet<Socket>()
+    private var requestBarrier: CountDownLatch? = null
+    private var deliveryGate: CountDownLatch? = null
     private var bytes = ByteArray(256 * 1024) { (it % 251).toByte() }
 
     @Before fun setUp() {
@@ -44,23 +54,38 @@ class EpicBoundedInstallTest {
                 val socket = try { server.accept() } catch (e: SocketException) {
                     if (server.isClosed) break else throw e
                 }
-                socket.use {
-                    it.soTimeout = 5000
-                    val reader = it.getInputStream().bufferedReader()
-                    while (!reader.readLine().isNullOrEmpty()) { }
-                    requests.incrementAndGet()
-                    val out = it.getOutputStream()
-                    if (reject.get()) {
-                        out.write("HTTP/1.1 503 Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".toByteArray())
-                    } else {
-                        val header = ByteBuffer.allocate(41).order(ByteOrder.LITTLE_ENDIAN)
-                            .putInt(0xB1FE3AA2.toInt()).putInt(1).putInt(41).putInt(bytes.size)
-                            .put(ByteArray(24)).put(0).array()
-                        out.write("HTTP/1.1 200 OK\r\nContent-Length: ${header.size + bytes.size}\r\nConnection: close\r\n\r\n".toByteArray())
-                        out.write(header)
-                        out.write(bytes)
-                    }
-                    out.flush()
+                sockets.add(socket)
+                handlers.submit {
+                    try {
+                        socket.use {
+                            it.soTimeout = 5000
+                            val reader = it.getInputStream().bufferedReader()
+                            val request = reader.readLine()
+                            while (!reader.readLine().isNullOrEmpty()) { }
+                            requests.incrementAndGet()
+                            requestBarrier?.let { barrier ->
+                                barrier.countDown()
+                                check(barrier.await(5, TimeUnit.SECONDS)) { "Requests did not overlap" }
+                            }
+                            val out = it.getOutputStream()
+                            val rejectedGuid = rejectGuid.get()
+                            if (reject.get() || (rejectedGuid != 0 && request.contains("_%08X".format(rejectedGuid)))) {
+                                out.write("HTTP/1.1 503 Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".toByteArray())
+                            } else {
+                                val header = ByteBuffer.allocate(41).order(ByteOrder.LITTLE_ENDIAN)
+                                    .putInt(0xB1FE3AA2.toInt()).putInt(1).putInt(41).putInt(bytes.size)
+                                    .put(ByteArray(24)).put(0).array()
+                                out.write("HTTP/1.1 200 OK\r\nContent-Length: ${header.size + bytes.size}\r\nConnection: close\r\n\r\n".toByteArray())
+                                out.write(header)
+                                out.flush()
+                                deliveryGate?.await(5, TimeUnit.SECONDS)
+                                out.write(bytes)
+                            }
+                            out.flush()
+                        }
+                    } catch (_: java.io.IOException) {
+                        // Cancellation disconnects an owned request while this fixture serves it.
+                    } finally { sockets.remove(socket) }
                 }
             }
         }.apply { start() }
@@ -78,7 +103,15 @@ class EpicBoundedInstallTest {
         }
     }
 
-    @After fun tearDown() { server.close(); worker.join(6000); assertFalse(worker.isAlive) }
+    @After fun tearDown() {
+        deliveryGate?.countDown()
+        server.close()
+        worker.join(6000)
+        sockets.forEach { it.close() }
+        handlers.shutdownNow()
+        assertTrue(handlers.awaitTermination(6, TimeUnit.SECONDS))
+        assertFalse(worker.isAlive)
+    }
 
     private fun sha(data: ByteArray) = MessageDigest.getInstance("SHA-1").digest(data)
 
@@ -159,13 +192,9 @@ class EpicBoundedInstallTest {
         }
         manifest.uniqueChunks.add(secondChunk)
         manifest.files.last().parts.single().guid = secondChunk.guid
-        val callback = object : EpicDownloadManager.Callback {
-            override fun onProgress(message: String, pct: Int) {}
-            override fun onStage(stage: String, done: Long, total: Long, items: Int, itemsTotal: Int) {
-                if (stage == "stream" && items == 1) reject.set(true)
-            }
-        }
-        assertThrows(EpicDownloadManager.InstallException::class.java) { install(callback = callback) }
+        // Prefetch may fail early, but replacement is still ordered by the writer.
+        rejectGuid.set(2)
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
         assertArrayEquals(bytes.copyOfRange(0, bytes.size / 2), File(root, "game.bin").readBytes())
         assertArrayEquals(byteArrayOf(6, 5, 4), File(root, "game.bin.part").readBytes())
         assertEquals(StoreGameSidecar.STATE_UPDATING, StoreGameSidecar.read(root)?.state)
@@ -376,7 +405,7 @@ class EpicBoundedInstallTest {
         assertTrue(samples.all { it.second == bytes.size.toLong() })
     }
 
-    @Test fun evictedSharedChunksAdvanceWriteProgressOnEveryRefetchWithoutChangingStages() {
+    @Test fun oversizedSharedChunkStaysPinnedAcrossPrefetchedFilesWithoutChangingStages() {
         bytes = ByteArray((EpicDownloadManager.MAX_RETAINED_CHUNK_CACHE_BYTES + 1024).toInt()) { (it % 251).toByte() }
         manifest.uniqueChunks.single().apply {
             windowSize = bytes.size
@@ -407,7 +436,7 @@ class EpicBoundedInstallTest {
             }
         }
         assertNotNull(install(callback = callback))
-        assertEquals(3, requests.get())
+        assertEquals(1, requests.get())
         assertEquals(listOf(0L, 1024L, 2048L, 3072L), samples)
         assertEquals("install", stages.last())
         repeat(3) { assertArrayEquals(bytes.copyOfRange(it * 1024, (it + 1) * 1024), File(root, "part-$it.bin").readBytes()) }
@@ -442,4 +471,175 @@ class EpicBoundedInstallTest {
         assertEquals(2 * half, EpicDownloadManager.estimateStagedFilePeak(manifest.files, emptyMap()))
         assertEquals(bytes.size.toLong(), EpicDownloadManager.estimatePeakAdditionalBytes(manifest.files, chunks, emptyMap(), false, old))
     }
+
+    private fun distinctFiles(count: Int) {
+        val template = manifest.uniqueChunks.single()
+        manifest.uniqueChunks = (1..count).map { id ->
+            EpicDownloadManager.ChunkInfo().apply {
+                guid = intArrayOf(id, 0, 0, 0)
+                windowSize = template.windowSize
+                fileSize = template.fileSize
+                sha1 = template.sha1
+            }
+        }
+        manifest.files = manifest.uniqueChunks.mapIndexed { index, chunk ->
+            gameFile("file-$index.bin", 0, bytes.size).apply { parts.single().guid = chunk.guid }
+        }
+    }
+
+    @Test fun prefetchOverlapsEightSinglePartFilesButNeverFetchesTheWholeManifest() {
+        distinctFiles(20)
+        requestBarrier = CountDownLatch(8)
+        val cancel = AtomicBoolean()
+        var sawWindow = false
+        assertNull(install(cancel, EpicDownloadManager.Callback { message, _ ->
+            if (message.startsWith("Downloading chunks (")) {
+                assertEquals(8, requests.get())
+                val cacheBytes = File(root, ".chunks").listFiles()!!.sumOf { it.length() }
+                assertTrue(cacheBytes <= 8L * bytes.size)
+                sawWindow = true
+                cancel.set(true)
+            }
+        }))
+        assertTrue("An eight-request barrier cannot pass with a sequential downloader", sawWindow)
+        assertEquals(8, requests.get())
+        assertTrue(root.walk().none { it.name.endsWith(".stage") || it.name.endsWith(".part") })
+    }
+
+    @Test fun parallelWindowSpaceThresholdIsCheckedBeforeStartingAnyRequest() {
+        distinctFiles(20)
+        val cache = temp.newFolder("parallel-scratch")
+        val peak = EpicDownloadManager.estimatePeakAdditionalBytes(
+            manifest.files, manifest.uniqueChunks.associateBy { it.guidStr() }, emptyMap(), false,
+        )
+        assertEquals(8L * bytes.size, peak)
+        ShadowStatFs.registerStats(cache.path, 1024, peak.toInt() / 4096 - 1, peak.toInt() / 4096 - 1)
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install(cache = cache.path) }
+        assertEquals(0, requests.get())
+        ShadowStatFs.registerStats(cache.path, 1024, peak.toInt() / 4096, peak.toInt() / 4096)
+        requestBarrier = CountDownLatch(8)
+        assertNotNull(install(cache = cache.path))
+        assertEquals(20, requests.get())
+        manifest.files.forEach { assertArrayEquals(bytes, File(root, it.filename).readBytes()) }
+    }
+
+    @Test fun cancellationDuringParallelNetworkReadsJoinsWorkersBeforeReturning() {
+        distinctFiles(20)
+        requestBarrier = CountDownLatch(8)
+        deliveryGate = CountDownLatch(1)
+        val cancel = AtomicBoolean()
+        val outcome = java.util.concurrent.atomic.AtomicReference<Any?>()
+        val installer = Thread {
+            try { outcome.set(install(cancel)) } catch (e: Throwable) { outcome.set(e) }
+        }.apply { start() }
+        try {
+            assertTrue(requestBarrier!!.await(5, TimeUnit.SECONDS))
+            cancel.set(true)
+            installer.join(10_000)
+            assertFalse("Installer must disconnect and join all download workers", installer.isAlive)
+            assertNull(outcome.get())
+            val cache = File(root, ".chunks")
+            val snapshot = cache.listFiles()?.associate { it.name to it.length() }.orEmpty()
+            deliveryGate!!.countDown()
+            Thread.sleep(150)
+            assertEquals(snapshot, cache.listFiles()?.associate { it.name to it.length() }.orEmpty())
+            assertTrue(snapshot.keys.none { it.endsWith(".part") })
+            assertTrue(root.walk().none { it.name.endsWith(".stage") })
+            assertEquals(8, requests.get())
+        } finally {
+            cancel.set(true)
+            deliveryGate!!.countDown()
+            installer.join(10_000)
+        }
+    }
+
+    @Test fun failedParallelFetchJoinsOtherNetworkReadersBeforeReturning() {
+        distinctFiles(20)
+        requestBarrier = CountDownLatch(8)
+        deliveryGate = CountDownLatch(1)
+        rejectGuid.set(1)
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+        assertEquals(8, requests.get())
+        val cache = File(root, ".chunks")
+        val snapshot = cache.listFiles()?.associate { it.name to it.length() }.orEmpty()
+        deliveryGate!!.countDown()
+        Thread.sleep(150)
+        assertEquals(snapshot, cache.listFiles()?.associate { it.name to it.length() }.orEmpty())
+        assertTrue(snapshot.keys.none { it.endsWith(".part") })
+        assertTrue(root.walk().none { it.name.endsWith(".stage") })
+    }
+
+    @Test fun ownedOrphanStagingIsRecoveredBeforeTheSpaceProbeWithoutTouchingManifestFiles() {
+        val staging = EpicDownloadManager.recoverOwnedStaging(root, manifest.files)
+        File(staging, "native-123-0.stage").writeBytes(ByteArray(bytes.size))
+        val legitimate = File(root, ".droiddeck-epic-old.stage").apply { writeText("keep") }
+        oldInstall()
+        val peak = 2 * bytes.size
+        ShadowStatFs.registerStats(root.path, 1024, peak / 4096, peak / 4096)
+        assertNotNull(install())
+        assertEquals("keep", legitimate.readText())
+        assertEquals(listOf("owner"), staging.listFiles()!!.map { it.name })
+    }
+
+    @Test fun stagingRecoveryRefusesUnownedAndManifestOwnedNamespaces() {
+        val staging = File(root, EpicDownloadManager.STAGING_DIRECTORY).apply { mkdir() }
+        val untouched = File(staging, "legitimate.stage").apply { writeText("keep") }
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+        assertEquals("keep", untouched.readText())
+        File(staging, "owner").writeText("DroidDeck Epic staging v1\n")
+        manifest.files.first().filename = "${EpicDownloadManager.STAGING_DIRECTORY}/legitimate.stage"
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+        assertEquals("keep", untouched.readText())
+        assertEquals(0, requests.get())
+    }
+
+    @Test fun loadedNativeAbiFailureIsExplicitAndNeverFallsBackToJavaDownloads() {
+        val field = com.droiddeck.launcher.stores.StoresNative::class.java.getDeclaredField("loaded")
+        field.isAccessible = true
+        field.set(null, true)
+        try {
+            oldInstall()
+            val error = assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+            assertTrue(error.message.orEmpty().contains("Native Epic install failed"))
+            assertEquals(0, requests.get())
+            assertOldInstallIntact()
+        } finally {
+            com.droiddeck.launcher.stores.StoresNative.forget()
+        }
+    }
+
+    @Test fun stagingRecoveryRejectsSymlinkNamespaceOwnerAndEntriesWithoutDeletingOutsideData() {
+        val outside = temp.newFolder("outside-staging")
+        val externalStage = File(outside, "keep.stage").apply { writeText("keep") }
+        val externalOwner = File(outside, "owner").apply { writeText("DroidDeck Epic staging v1\n") }
+        val staging = File(root, EpicDownloadManager.STAGING_DIRECTORY)
+        try {
+        java.nio.file.Files.createSymbolicLink(staging.toPath(), outside.toPath())
+        } catch (e: java.nio.file.FileSystemException) {
+        org.junit.Assume.assumeNoException("Host does not permit symbolic links", e)
+        }
+        try {
+        val error = assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+        assertTrue(error.message.orEmpty().contains("Unsafe Epic staging"))
+        assertEquals("keep", externalStage.readText())
+        java.nio.file.Files.delete(staging.toPath())
+        staging.mkdir()
+        java.nio.file.Files.createSymbolicLink(File(staging, "owner").toPath(), externalOwner.toPath())
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+        java.nio.file.Files.delete(File(staging, "owner").toPath())
+        File(staging, "owner").writeText("DroidDeck Epic staging v1\n")
+        val localStage = File(staging, "local.stage").apply { writeText("local") }
+        java.nio.file.Files.createSymbolicLink(File(staging, "linked.stage").toPath(), externalStage.toPath())
+        assertThrows(EpicDownloadManager.InstallException::class.java) { install() }
+        assertEquals("local", localStage.readText())
+        assertEquals("keep", externalStage.readText())
+        assertEquals(0, requests.get())
+        } finally {
+        if (java.nio.file.Files.isSymbolicLink(staging.toPath())) java.nio.file.Files.delete(staging.toPath())
+        for (name in listOf("owner", "linked.stage")) {
+            val entry = File(staging, name).toPath()
+            if (java.nio.file.Files.isSymbolicLink(entry)) java.nio.file.Files.delete(entry)
+        }
+        }
+    }
 }

app/src/test/java/com/droiddeck/launcher/stores/epic/EpicJsonManifestTest.kt

@@ -15,14 +15,14 @@ import org.robolectric.annotation.Config
 @Config(sdk = [33])
 class EpicJsonManifestTest {
     @Test
-    fun boundedInstallPeakCountsOnlyOneLiveChunkForSequentialChunks() {
+    fun boundedInstallPeakReservesParallelIncomingChunksBeforeWriting() {
         val first = chunk(1)
         val second = chunk(2)
         val file = file(part(1, 0, 100), part(2, 0, 100))
         val chunks = mapOf(first.guidStr() to first, second.guidStr() to second)
 
         assertEquals(300L, EpicDownloadManager.estimatePeakAdditionalBytes(listOf(file), chunks, emptyMap(), true))
-        assertEquals(100L, EpicDownloadManager.estimatePeakAdditionalBytes(listOf(file), chunks, emptyMap(), false))
+        assertEquals(200L, EpicDownloadManager.estimatePeakAdditionalBytes(listOf(file), chunks, emptyMap(), false))
     }
 
     @Test
@@ -45,7 +45,7 @@ class EpicJsonManifestTest {
     }
 
     @Test
-    fun distantSharedChunksAreEvictedAndRefetchedWithinTheCacheCeiling() {
+    fun sharedChunksWithinTheWindowStayPinnedWithinTheCacheCeiling() {
         val first = chunk(1).apply { windowSize = 40 * 1024 * 1024 }
         val second = chunk(2).apply { windowSize = 40 * 1024 * 1024 }
         val files = listOf(
@@ -63,6 +63,43 @@ class EpicJsonManifestTest {
         assertTrue(cachePeak <= EpicDownloadManager.MAX_RETAINED_CHUNK_CACHE_BYTES + 40L * 1024 * 1024)
     }
 
+    @Test fun scheduleDeduplicatesWindowAndEvictsOnlyUnpinnedReusableChunks() {
+        val size = 40L * 1024 * 1024
+        val chunks = (1..12).map { chunk(it).apply { windowSize = size.toInt() } }
+            .associateBy { it.guidStr() }
+        val files = (1..12).map { file(part(it, 0, 1)) } + (1..12).map { file(part(it, 0, 1)) }
+        val schedule = EpicDownloadManager.PrefetchSchedule(files, chunks, emptyMap())
+        assertEquals(chunks.keys.take(8), schedule.fill())
+        assertEquals(emptyList<String>(), schedule.fill())
+        assertEquals(8 * size, schedule.bytes())
+        assertEquals(emptyList<String>(), schedule.consume())
+        assertEquals(listOf(chunks.keys.elementAt(8)), schedule.fill())
+        assertEquals(9 * size, schedule.bytes())
+        assertEquals(listOf(chunks.keys.first()), schedule.consume())
+        var refetchedFirst = false
+        repeat(22) {
+            val fetched = schedule.fill()
+            if (chunks.keys.first() in fetched) refetchedFirst = true
+            assertTrue(schedule.bytes() <= EpicDownloadManager.MAX_RETAINED_CHUNK_CACHE_BYTES + 8 * size)
+            val released = schedule.consume()
+            assertTrue(released.none { it in schedule.window() })
+        }
+        assertTrue(refetchedFirst)
+        assertEquals(0L, schedule.bytes())
+        assertEquals(9 * size, EpicDownloadManager.estimatePeakAdditionalBytes(files, chunks, emptyMap(), false))
+    }
+
+    @Test fun cachedAndOldOutputBytesAreCreditedOnlyWhenTheScheduleReleasesThem() {
+        val chunks = (1..8).map { chunk(it) }.associateBy { it.guidStr() }
+        val files = (1..8).map { id -> file(part(id, 0, 100)).apply { filename = "file-$id" } }
+        val existing = chunks.keys.associateWith { 100L }
+        val old = files.associate { it.filename to 100L }
+        assertEquals(800L, EpicDownloadManager.estimatePeakAdditionalBytes(files, chunks, emptyMap(), false))
+        assertEquals(0L, EpicDownloadManager.estimatePeakAdditionalBytes(files, chunks, existing, false))
+        assertEquals(900L, EpicDownloadManager.estimatePeakAdditionalBytes(files, chunks, emptyMap(), true, old))
+        assertEquals(100L, EpicDownloadManager.estimatePeakAdditionalBytes(files, chunks, existing, true, old))
+    }
+
     @Test
     fun resumeCacheDropsCorruptAndUnneededEntriesButKeepsVerifiedChunks() {
         val dir = java.io.File(System.getProperty("user.dir"), "build/epic-cache-test-${System.nanoTime()}")

docs/development/stores.md

@@ -45,13 +45,46 @@ with a placeholder job. Failures are kept in the Downloads UI, notification and
 duplicated as transient failure toasts. Active queue persistence and ordinary backend pause/retry
 are unchanged.
 
-Epic installs use the same bounded Java writer whether the native engine is available or not.
-It fetches one chunk at a time, verifies it, and writes the required parts into the current
-game file. Chunks are deleted after their last use; reusable chunks are retained up to 64 MiB,
-with at most one additional incoming chunk before eviction. Evicted shared chunks may be
-downloaded again. This trades parallel-download throughput and potentially extra network
-traffic for bounded temporary disk usage. The native batch APIs remain available but are not
-used by this install path.
+Epic installs use **native adaptive fetch/assembly** when `libdroiddeckstores.so` loads.
+The Java delta pass sends a versioned snapshot of pending files, chunk metadata, verified cache
+lengths, old output lengths and both measured free-space probes to Rust. The native planner and
+execution share one ordered moving-window policy. Before any chunk request, Rust chooses the
+largest safe lookahead (128, 64, 32, 16, 8, 4, 2 or 1 part references) and reusable-cache budget
+(64 MiB, 16 MiB or zero) whose simulated staged-file and cache peaks fit.
+Native device IDs also detect an explicitly named cache on the game volume, so two separate
+paths cannot incorrectly turn one filesystem's requirement into two independent checks.
+
+| Storage situation | Native policy |
+| --- | --- |
+| Plenty of room | Up to 128 references ahead, up to 32 network requests, 64 MiB reusable data |
+| Tight | Smaller lookahead and/or reusable budget; parallel fetching remains where space permits |
+| Barely fits | One-reference lookahead, one request, no reusable retention if needed |
+| Does not fit even the minimal schedule | Explicit preflight error before requests or replacement |
+| Free space unknown | Conservative bounded lookahead (at most eight), not an invented zero |
+
+The native fetch coordinator reuses `fetch_core`'s host scheduling, retries, adaptive network
+window and decode pool. The staged writer runs concurrently, waiting only for its next required
+chunk, and reads parts through the native assembler's streaming copy primitive. GUIDs in the
+lookahead are reserved once before submission, at their full decoded size. Chunks are released
+at last use on **both** cache layouts; distant reusable data is oldest-first bounded and may
+be refetched. At most the selected reusable budget plus distinct lookahead decoded sizes is
+resident/reserved. New request admission re-probes both volumes and can shrink unsent lookahead
+or retention if headroom disappears; already in-flight reservations remain protected.
+If even existing reservations cannot fit, the run fails and joins its workers rather than
+waiting indefinitely or deleting old game files. An install to SD retains its internal scratch
+cache, so assembled output is the only game data written to the card.
+
+Only failure to **load** the native library selects the Java fallback. An ABI mismatch, native
+planning/network/hash/I/O failure or unsupported atomic swap does not silently rerun Java.
+Eight fallback Java workers prefetch the next eight chunk-part references, across file boundaries,
+while one writer assembles and verifies files in manifest order. Shared GUIDs within that window
+are fetched only once. Chunks are deleted after their last use; other reusable chunks are retained
+up to 64 MiB. At most eight additional decoded chunks (complete or in flight) coexist with that
+retained cache. The size bound is 64 MiB plus the decoded sizes of the window's distinct GUIDs,
+not a fixed eight-MiB allowance: manifests can supply larger chunk windows. Evicted distant shared
+chunks may be downloaded again. No worker downloads outside the moving window; cancellation and
+failure disconnect and join all workers before returning or removing the cache.
+The legacy native batch APIs remain available but are not used by the production install path.
 
 The earlier batch design intentionally prioritizes parallel-download throughput. Native
 `epic/assemble.rs::build_assemble_plan` enables `scratch_cache` only for a nonempty cache path.
@@ -61,9 +94,9 @@ when that flag is set**; the default beside-game cache is removed at the end. Th
 the scratch-cache assembly tests cover that distinction. Extending last-use reclamation to
 internal native assembly could keep parallel batch throughput and reduce the overlapping
 cache/output peak, but batch fetching still fills
-the cache before assembly. That is not equivalent to the bounded sequential writer's fixed
-retained-cache ceiling. This change keeps the bounded path and its explicit throughput/refetch
-tradeoff; it does not redesign or benchmark native download throughput.
+the cache before assembly. Production native adaptive and Java fallback paths instead overlap
+network fetches with assembly without retaining a whole-game cache. No native or Java throughput
+benchmark is claimed.
 
 The bounded Epic download bar, measured rate and ETA count pending file bytes written, not unique
 compressed chunks fetched. Shared chunks therefore cannot complete the bar before later files
@@ -71,9 +104,12 @@ are written, and an evicted chunk's refetch is reflected in the time spent doing
 Streaming stays in the Download stage until all files are written and verified, then enters
 Install for registration; it does not bounce between stages for each file.
 
-The space check follows the manifest delta pass and simulates file writes, chunk fetches,
-and evictions, rather than adding the entire cache to the entire game. Each file is written to a
-unique sibling staging file on its output volume (never the literal `<filename>.part`, which can
+The space check follows the manifest delta pass and uses the same pure moving-window policy as
+execution, reserving all incoming decoded chunk sizes before writing parts and simulating
+last-use releases and unpinned oldest-first evictions. Existing verified cache bytes count against
+current free space and are credited only at their scheduled release. This avoids adding the entire
+cache to the entire game without undercounting concurrent downloads. Each file is written to a
+unique file in an owned staging namespace on its output volume (never the literal `<filename>.part`, which can
 be a legitimate manifest file), size/hash verified, synced and atomically renamed over its output.
 The old file remains until that swap succeeds. The planner credits its bytes only after that
 file's complete staging write, not by subtracting all old outputs up front. Both volume checks
@@ -84,8 +120,17 @@ verified cached chunks are accounted for on resume. On a card, game and cache re
 are checked separately. File SHA-1 hashes are checked where supplied by the manifest; failed
 or cancelled staging files are removed without deleting old outputs. A successful run removes the
 cache. A resume within a large incomplete file restarts that file, not its individual parts.
-Abrupt process termination can leave a unique staging file; unlike normal failure/cancellation,
-its cleanup is not guaranteed, and the next space probe includes its occupied bytes.
+Both writers use `<installDir>/.droiddeck-epic-staging/`, whose `owner` marker identifies this
+namespace. Before the delta pass and free-space probes, the manager removes orphan `.stage`
+entries only from this owned namespace. A manifest naming that directory or any descendant
+fails without deletion; an unowned directory or unexpected entry also fails conservatively.
+The namespace must be a real directory at the expected canonical child of the install root.
+Ownership markers and staging entries must be no-follow regular files; symlink directories,
+markers and entries are rejected before recovery deletes anything.
+Ordinary game files with `.stage` or `.part` suffixes elsewhere are never recovery targets.
+Abrupt process termination is therefore recoverable on the next run without inferring ownership
+from a loose filename prefix. Cancellation/failure waits for all native/Java fetch writers before
+returning; completed verified cache chunks remain usable on resume.
 
 ## The sidecar