diff --git a/app/src/main/java/com/droiddeck/launcher/stores/epic/EpicDownloadManager.java b/app/src/main/java/com/droiddeck/launcher/stores/epic/EpicDownloadManager.java index c5240f4f1..d0ea636cf 100644 --- a/app/src/main/java/com/droiddeck/launcher/stores/epic/EpicDownloadManager.java +++ b/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 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 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 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 reservedPaths) throws IOException { + static File recoverOwnedStaging(File installDir, List 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 files, + Map chunks, Map cached, Map 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 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 ".part". + // Owned staging is on the output volume, never a manifest's literal ".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 parts = new ArrayList<>(); + final Map resident; + final Map refs; + final Map chunks; + int cursor; + + PrefetchSchedule(List files, Map chunks, Map 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 window() { + return new LinkedHashSet<>(parts.subList(cursor, Math.min(parts.size(), cursor + PREFETCH_PARTS))); + } + + List fill() { + List 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 consume() { + List release = new ArrayList<>(); + String guid = parts.get(cursor++); + if (consumeChunkReference(refs, guid)) { + resident.remove(guid); + release.add(guid); + } + Set pinned = window(); + long reusable = 0; + for (Map.Entry entry : resident.entrySet()) { + if (!pinned.contains(entry.getKey())) reusable += entry.getValue(); + } + Iterator> it = resident.entrySet().iterator(); + while (reusable > MAX_RETAINED_CHUNK_CACHE_BYTES && it.hasNext()) { + Map.Entry 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> jobs = new LinkedHashMap<>(); + final Set connections = ConcurrentHashMap.newKeySet(); + final String chunkDir; + final List cdns; + + ChunkPrefetch(List files, Map chunks, Map cached, + File cacheDir, String chunkDir, List 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 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 files, Map chunksByGuid, Map cachedChunkBytes, boolean includeFileWrites, Map oldFileBytes) { - Map refs = chunkReferences(files); - Map 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 cdnUrls, File outFile, AtomicBoolean cancel) { + private static boolean downloadChunkStreaming(ChunkInfo chunk, String chunkDir, List cdnUrls, + File outFile, AtomicBoolean cancel, AtomicBoolean stopped, + Set 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(); diff --git a/app/src/main/java/com/droiddeck/launcher/stores/epic/EpicNative.kt b/app/src/main/java/com/droiddeck/launcher/stores/epic/EpicNative.kt index 57f268170..eca9c973e 100644 --- a/app/src/main/java/com/droiddeck/launcher/stores/epic/EpicNative.kt +++ b/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 `/.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(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, pendingFileIdx: IntArray, expectedChunks: Int, expectedBytes: Long, diff --git a/app/src/main/rust/stores/JNI.md b/app/src/main/rust/stores/JNI.md index 9f1b38550..2ba742eed 100644 --- a/app/src/main/rust/stores/JNI.md +++ b/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 diff --git a/app/src/main/rust/stores/src/epic.rs b/app/src/main/rust/stores/src/epic.rs index 258c273b9..48a4befff 100644 --- a/app/src/main/rust/stores/src/epic.rs +++ b/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 `/.chunks/`), 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; diff --git a/app/src/main/rust/stores/src/epic/adaptive.rs b/app/src/main/rust/stores/src/epic/adaptive.rs new file mode 100644 index 000000000..7113723ea --- /dev/null +++ b/app/src/main/rust/stores/src/epic/adaptive.rs @@ -0,0 +1,1585 @@ +//! Budgeted native install: ordered lookahead, shared peak schedule, fetch-core overlap and +//! verified per-file atomic swaps. The JNI snapshot is the Java delta pass's exact pending set. + +use crate::fetch_core::{run_fetch, FetchItem, FetchOptions, FetchSink, SinkError}; +use serde_json::Value; +use sha1::{Digest, Sha1}; +use std::collections::{HashMap, HashSet}; +use std::fs::{self, OpenOptions}; +use std::io::{BufWriter, Write}; +use std::path::{Component, Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Condvar, Mutex}; +use std::time::Duration; + +const MIB: u64 = 1024 * 1024; +const RETAINED: u64 = 64 * MIB; + +#[derive(Clone, Debug)] +pub struct Chunk { + guid: String, + path: String, + size: u64, + wire: u64, + sha: Option<[u8; 20]>, + cached: Option, + exact: bool, +} +#[derive(Clone, Debug)] +pub struct Part { + chunk: usize, + offset: u64, + size: u64, +} +#[derive(Clone, Debug)] +pub struct Output { + path: String, + sha: Option<[u8; 20]>, + old: u64, + parts: Vec, +} +#[derive(Clone, Debug)] +pub struct Request { + pub install: PathBuf, + pub cache: PathBuf, + pub staging: PathBuf, + pub same_volume: bool, + pub free_game: Option, + pub free_cache: Option, + pub chunks: Vec, + pub files: Vec, + pub hosts: Vec, + pub max_workers: usize, +} + +fn uint(v: &Value, key: &str) -> Result { + v[key].as_u64().ok_or_else(|| format!("invalid {key}")) +} +fn text(v: &Value, key: &str) -> Result { + v[key] + .as_str() + .map(str::to_string) + .ok_or_else(|| format!("invalid {key}")) +} +fn digest(v: &Value) -> Result, String> { + let s = v.as_str().ok_or("invalid sha1")?; + if s.is_empty() { + return Ok(None); + } + if s.len() != 40 || !s.is_ascii() { + return Err("invalid sha1".into()); + } + let mut out = [0; 20]; + for (i, b) in out.iter_mut().enumerate() { + *b = u8::from_str_radix(&s[i * 2..i * 2 + 2], 16).map_err(|_| "invalid sha1")?; + } + Ok(out.iter().any(|b| *b != 0).then_some(out)) +} +fn relative(s: &str) -> Result<(), String> { + if s.is_empty() + || s.contains('\\') + || s.contains(':') + || Path::new(s) + .components() + .any(|c| !matches!(c, Component::Normal(_))) + { + return Err(format!("unsafe manifest path {s}")); + } + Ok(()) +} +impl Request { + pub fn parse(json: &str) -> Result { + let v: Value = serde_json::from_str(json).map_err(|e| e.to_string())?; + if uint(&v, "schema_version")? != 1 { + return Err("unsupported adaptive request schema".into()); + } + let mut chunks = Vec::new(); + let mut by_guid = HashMap::new(); + for c in v["chunks"].as_array().ok_or("missing chunks")? { + let guid = text(c, "guid")?; + if guid.len() != 32 + || !guid.bytes().all(|b| b.is_ascii_hexdigit()) + || by_guid.contains_key(&guid) + { + return Err("invalid or duplicate chunk GUID".into()); + } + let path = text(c, "path")?; + relative(&path)?; + by_guid.insert(guid.clone(), chunks.len()); + chunks.push(Chunk { + guid, + path, + size: uint(c, "size")?, + wire: uint(c, "wire_size")?, + sha: digest(&c["sha1"])?, + cached: c["cached"].as_u64(), + exact: c["exact_size"].as_bool().ok_or("missing exact_size")?, + }); + } + let mut files = Vec::new(); + for f in v["files"].as_array().ok_or("missing files")? { + let path = text(f, "path")?; + relative(&path)?; + if path.split('/').next() == Some(".droiddeck-epic-staging") { + return Err("manifest conflicts with owned staging".into()); + } + let mut parts = Vec::new(); + for p in f["parts"].as_array().ok_or("missing parts")? { + let chunk = *by_guid + .get(&text(p, "guid")?) + .ok_or("missing chunk metadata")?; + let offset = uint(p, "offset")?; + let size = uint(p, "size")?; + if offset + .checked_add(size) + .filter(|n| *n <= chunks[chunk].size) + .is_none() + { + return Err("part exceeds decoded window".into()); + } + parts.push(Part { + chunk, + offset, + size, + }); + } + files.push(Output { + path, + parts, + sha: digest(&f["sha1"])?, + old: uint(f, "old_size")?, + }); + } + let install = PathBuf::from(text(&v, "install_dir")?); + let staging = PathBuf::from(text(&v, "staging_dir")?); + Self::validate_staging(&install, &staging)?; + let hosts = v["hosts"] + .as_array() + .ok_or("missing hosts")? + .iter() + .map(|h| { + h.as_str() + .map(str::to_string) + .ok_or_else(|| "invalid host".to_string()) + }) + .collect::, _>>()?; + if hosts.is_empty() { + return Err("no Epic CDN".into()); + } + let cache = PathBuf::from(text(&v, "cache_dir")?); + let declared_same = v["same_volume"].as_bool().ok_or("missing same_volume")?; + let same_volume = declared_same || Self::same_filesystem(&install, &cache); + let game_probe = v["free_game"].as_u64(); + let cache_probe = v["free_cache"].as_u64(); + let free_game = if same_volume { + match (game_probe, cache_probe) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + } + } else { + game_probe + }; + Ok(Self { + install, + staging, + cache, + chunks, + files, + hosts, + same_volume, + free_game, + free_cache: cache_probe, + max_workers: uint(&v, "max_workers")?.clamp(1, 32) as usize, + }) + } + + fn validate_staging(install: &Path, staging: &Path) -> Result<(), String> { + let expected = fs::canonicalize(install) + .map_err(|e| e.to_string())? + .join(".droiddeck-epic-staging"); + let metadata = fs::symlink_metadata(staging).map_err(|e| e.to_string())?; + if !metadata.file_type().is_dir() + || fs::canonicalize(staging).map_err(|e| e.to_string())? != expected + { + return Err("unsafe staging directory; refusing to traverse it".into()); + } + let owner = staging.join("owner"); + let metadata = fs::symlink_metadata(&owner).map_err(|e| e.to_string())?; + if !metadata.file_type().is_file() + || metadata.len() != 26 + || fs::canonicalize(&owner).map_err(|e| e.to_string())? != expected.join("owner") + || fs::read(&owner).map_err(|e| e.to_string())? != b"DroidDeck Epic staging v1\n" + { + return Err("unowned staging directory".into()); + } + for entry in fs::read_dir(staging).map_err(|e| e.to_string())? { + let entry = entry.map_err(|e| e.to_string())?; + let name = entry.file_name(); + if name == "owner" { + continue; + } + if !entry.file_type().map_err(|e| e.to_string())?.is_file() + || !name.to_string_lossy().ends_with(".stage") + || fs::canonicalize(entry.path()).map_err(|e| e.to_string())? + != expected.join(&name) + { + return Err("unsafe staging entry".into()); + } + } + Ok(()) + } + + #[cfg(unix)] + fn same_filesystem(a: &Path, b: &Path) -> bool { + use std::os::unix::fs::MetadataExt; + match (fs::metadata(a), fs::metadata(b)) { + (Ok(a), Ok(b)) => a.dev() == b.dev(), + _ => false, + } + } + #[cfg(not(unix))] + fn same_filesystem(_a: &Path, _b: &Path) -> bool { + false + } + fn total(&self) -> u64 { + self.files + .iter() + .flat_map(|f| &f.parts) + .map(|p| p.size) + .sum() + } +} + +/// Resident includes full-size reservations for in-flight writes; only ready chunks can be read. +#[derive(Clone, Debug)] +struct Schedule { + order: Vec, + cursor: usize, + refs: Vec, + resident: Vec>, + oldest: Vec, + lookahead: usize, + retained: u64, +} +impl Schedule { + fn new(req: &Request, lookahead: usize, retained: u64) -> Self { + let order: Vec<_> = req + .files + .iter() + .flat_map(|f| f.parts.iter().map(|p| p.chunk)) + .collect(); + let mut refs = vec![0; req.chunks.len()]; + for &c in &order { + refs[c] += 1; + } + let resident: Vec<_> = req.chunks.iter().map(|c| c.cached).collect(); + let oldest = (0..resident.len()) + .filter(|&c| resident[c].is_some()) + .collect(); + Self { + order, + cursor: 0, + refs, + resident, + oldest, + lookahead, + retained, + } + } + fn window(&self) -> HashSet { + self.order[self.cursor..self.order.len().min(self.cursor + self.lookahead)] + .iter() + .copied() + .collect() + } + fn fill(&mut self, req: &Request) -> Vec { + let mut fetch = Vec::new(); + for &c in &self.order[self.cursor..self.order.len().min(self.cursor + self.lookahead)] { + if self.resident[c].is_none() { + self.resident[c] = Some(req.chunks[c].size); + self.oldest.push(c); + fetch.push(c); + } + } + fetch + } + fn consume(&mut self) -> Vec { + let c = self.order[self.cursor]; + self.cursor += 1; + self.refs[c] -= 1; + let mut release = Vec::new(); + if self.refs[c] == 0 { + self.resident[c] = None; + release.push(c); + } + let pinned = self.window(); + let mut reusable: u64 = self + .oldest + .iter() + .filter(|c| !pinned.contains(c)) + .map(|&c| self.resident[c].unwrap_or(0)) + .sum(); + for &c in &self.oldest { + if reusable <= self.retained { + break; + } + if pinned.contains(&c) { + continue; + } + if let Some(bytes) = self.resident[c].take() { + reusable -= bytes; + release.push(c); + } + } + self.oldest.retain(|&c| self.resident[c].is_some()); + release + } + fn bytes(&self) -> u64 { + self.oldest + .iter() + .map(|&c| self.resident[c].unwrap_or(0)) + .sum() + } +} + +#[derive(Clone, Copy, Debug)] +pub struct Budget { + pub lookahead: usize, + pub retained: u64, + pub game_peak: u64, + pub cache_peak: u64, +} +fn peaks(req: &Request, lookahead: usize, retained: u64) -> (u64, u64) { + let mut s = Schedule::new(req, lookahead, retained); + let initial = s.bytes() as i128; + let mut game = 0i128; + let mut game_peak = 0i128; + let mut cache_peak = 0i128; + for file in &req.files { + for part in &file.parts { + s.fill(req); + let cache = s.bytes() as i128 - initial; + cache_peak = cache_peak.max(cache); + game_peak = game_peak.max(game + if req.same_volume { cache } else { 0 }); + game += part.size as i128; + game_peak = game_peak.max(game + if req.same_volume { cache } else { 0 }); + s.consume(); + } + game -= file.old as i128; + } + (game_peak.max(0) as u64, cache_peak.max(0) as u64) +} +fn fits(needed: u64, free: Option) -> bool { + free.is_none_or(|n| needed <= n) +} +pub fn choose_budget(req: &Request) -> Result { + // Unknown probes deliberately choose a conservative bounded mode, never "0 free". + let unknown = req.free_game.is_none() || (!req.same_volume && req.free_cache.is_none()); + for lookahead in [128, 64, 32, 16, 8, 4, 2, 1] { + if unknown && lookahead > 8 { + continue; + } + for retained in [RETAINED, 16 * MIB, 0] { + let (game_peak, cache_peak) = peaks(req, lookahead, retained); + if fits(game_peak, req.free_game) + && (req.same_volume || fits(cache_peak, req.free_cache)) + { + return Ok(Budget { + lookahead, + retained, + game_peak, + cache_peak, + }); + } + } + } + let (game, cache) = peaks(req, 1, 0); + Err(format!("Not enough free space for safe staged Epic replacement: minimum game additional={game}, cache additional={cache}; free game={:?}, cache={:?}", req.free_game, req.free_cache)) +} + +fn remaining_peaks( + req: &Request, + schedule: &Schedule, + fi: usize, + pi: usize, + disk_cache: u64, +) -> (u64, u64) { + let mut s = schedule.clone(); + let mut game = 0i128; + let mut game_peak = 0i128; + let mut cache_peak = 0i128; + let outstanding = s.bytes() as i128 - disk_cache as i128; + cache_peak = cache_peak.max(outstanding); + if req.same_volume { + game_peak = game_peak.max(outstanding); + } + for (index, file) in req.files.iter().enumerate().skip(fi) { + for part in file.parts.iter().skip(if index == fi { pi } else { 0 }) { + s.fill(req); + let cache = s.bytes() as i128 - disk_cache as i128; + cache_peak = cache_peak.max(cache); + game_peak = game_peak.max(game + if req.same_volume { cache } else { 0 }); + game += part.size as i128; + game_peak = game_peak.max(game + if req.same_volume { cache } else { 0 }); + s.consume(); + } + game -= file.old as i128; + } + (game_peak.max(0) as u64, cache_peak.max(0) as u64) +} + +#[cfg(unix)] +fn available(path: &Path) -> Option { + use std::os::unix::ffi::OsStrExt; + let path = std::ffi::CString::new(path.as_os_str().as_bytes()).ok()?; + let mut stat = std::mem::MaybeUninit::::uninit(); + if unsafe { libc::statvfs(path.as_ptr(), stat.as_mut_ptr()) } != 0 { + return None; + } + let stat = unsafe { stat.assume_init() }; + Some((stat.f_bavail as u64).saturating_mul(stat.f_frsize as u64)) +} +#[cfg(not(unix))] +fn available(_path: &Path) -> Option { + None +} + +/// Re-probe at admission boundaries. Already dispatched reservations cannot be withdrawn. +/// Shrink only unsent lookahead/reusable budget; never wait forever for unrelated space use. +fn live_admit( + req: &Request, + state: &mut State, + fi: usize, + pi: usize, + stage_remaining: u64, +) -> Result<(), String> { + let disk_cache = state + .schedule + .oldest + .iter() + .map(|&index| { + if state.ready[index] { + return state.disk_sizes[index]; + } + if !state.writing[index] { + return 0; + } + let c = &req.chunks[index]; + fs::metadata(req.cache.join(&c.guid)) + .map(|m| m.len()) + .unwrap_or(0) + + fs::metadata(super::chunk::part_path(&req.cache.join(&c.guid))) + .map(|m| m.len()) + .unwrap_or(0) + }) + .sum(); + let free_game = available(&req.install); + let free_cache = available(&req.cache); + admit_with_probes( + req, + state, + fi, + pi, + stage_remaining, + disk_cache, + free_game, + free_cache, + ) +} + +fn admit_with_probes( + req: &Request, + state: &mut State, + fi: usize, + pi: usize, + stage_remaining: u64, + disk_cache: u64, + free_game: Option, + free_cache: Option, +) -> Result<(), String> { + loop { + let incoming: u64 = state + .schedule + .window() + .iter() + .filter(|&&c| state.schedule.resident[c].is_none()) + .map(|&c| req.chunks[c].size) + .sum(); + let cache_now = state + .schedule + .bytes() + .saturating_add(incoming) + .saturating_sub(disk_cache); + let game_now = stage_remaining.saturating_add(if req.same_volume { cache_now } else { 0 }); + if fits(game_now, free_game) && (req.same_volume || fits(cache_now, free_cache)) { + return Ok(()); + } + // Full remaining simulation is only needed when live admission headroom is exhausted. + let (game, cache) = remaining_peaks(req, &state.schedule, fi, pi, disk_cache); + if fits(game, free_game) && (req.same_volume || fits(cache, free_cache)) { + return Ok(()); + } + if state.schedule.oldest.iter().any(|&c| !state.ready[c]) { + return Err("Not enough free space for already reserved Epic downloads".into()); + } + if state.schedule.retained > 0 { + state.schedule.retained = 0; + continue; + } + if state.schedule.lookahead > 1 { + state.schedule.lookahead /= 2; + continue; + } + return Err(format!("Not enough free space during Epic install: need additional game={game} cache={cache}, free game={free_game:?} cache={free_cache:?}")); + } +} + +struct State { + schedule: Schedule, + ready: Vec, + writing: Vec, + disk_sizes: Vec, + pending: Vec, + error: Option, + done: bool, +} +struct Shared { + state: Mutex, + changed: Condvar, + stop: AtomicBool, +} +struct Sink<'a> { + req: &'a Request, + shared: &'a Shared, +} +impl FetchSink for Sink<'_> { + fn process(&self, item: &FetchItem, body: Vec) -> Result { + if self.shared.stop.load(Ordering::Acquire) { + return Err(SinkError::Fatal("cancelled".into())); + } + let c = item.id as usize; + let chunk = &self.req.chunks[c]; + self.shared.state.lock().unwrap().writing[c] = true; + let written = super::chunk::write_verified_chunk_bounded( + &body, + chunk.sha.as_ref(), + &self.req.cache.join(&chunk.guid), + chunk.size, + chunk.exact, + ); + let mut state = self.shared.state.lock().unwrap(); + state.writing[c] = false; + let written = written.map_err(SinkError::Retry)?; + state.ready[c] = true; + state.disk_sizes[c] = written; + self.shared.changed.notify_all(); + Ok(chunk.wire) + } +} + +/// JNI owns this entire blocking run. All scoped workers join before completion is signalled. +pub fn run( + req: &Request, + budget: Budget, + cancel: &AtomicBool, + before_swap: &(dyn Fn() -> Result<(), String> + Sync), + progress: &(dyn Fn(u64, u64, u64, u64) + Sync), + log: &(dyn Fn(&str) + Sync), +) -> Result { + let shared = Shared { + state: Mutex::new(State { + schedule: Schedule::new(req, budget.lookahead, budget.retained), + ready: req.chunks.iter().map(|c| c.cached.is_some()).collect(), + writing: vec![false; req.chunks.len()], + disk_sizes: req.chunks.iter().map(|c| c.cached.unwrap_or(0)).collect(), + pending: Vec::new(), + error: None, + done: false, + }), + changed: Condvar::new(), + stop: AtomicBool::new(false), + }; + log(&format!( + "adaptive-native lookahead={} retained={} game_peak={} cache_peak={} workers={}", + budget.lookahead, + budget.retained, + budget.game_peak, + budget.cache_peak, + req.max_workers.min(budget.lookahead) + )); + let result = std::thread::scope(|scope| { + scope.spawn(|| { + crate::priority::background(); + let sink = Sink { + req, + shared: &shared, + }; + loop { + let batch = { + let mut state = shared.state.lock().unwrap(); + while state.pending.is_empty() + && !state.done + && !shared.stop.load(Ordering::Acquire) + { + state = shared + .changed + .wait_timeout(state, Duration::from_millis(100)) + .unwrap() + .0; + } + if state.done || shared.stop.load(Ordering::Acquire) { + break; + } + std::mem::take(&mut state.pending) + }; + let items = batch + .into_iter() + .map(|c| FetchItem { + id: c as u64, + urls: req + .hosts + .iter() + .map(|h| format!("{h}/{}", req.chunks[c].path)) + .collect(), + reserve: req.chunks[c].wire, + range: None, + }) + .collect(); + let opts = FetchOptions { + max_workers: req.max_workers.min(budget.lookahead), + per_host_cap: super::plan::per_host_cap( + req.max_workers.min(budget.lookahead), + req.hosts.len(), + ), + timeout: Duration::from_secs(60), + process_workers: 2, + headers: vec![("User-Agent".into(), super::USER_AGENT.into())], + label: "epic adaptive".into(), + ..Default::default() + }; + let fetched = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_fetch( + items, + &req.hosts, + &opts, + &sink, + &shared.stop, + &|_, _| {}, + log, + ) + })); + let outcome = match fetched { + Ok(outcome) => outcome, + Err(_) => { + let mut state = shared.state.lock().unwrap(); + state.error = Some("native fetch panicked".into()); + shared.stop.store(true, Ordering::Release); + shared.changed.notify_all(); + break; + } + }; + if !outcome.ok() { + let mut state = shared.state.lock().unwrap(); + state.error = Some(outcome.error.unwrap_or_else(|| "cancelled".into())); + shared.stop.store(true, Ordering::Release); + shared.changed.notify_all(); + break; + } + } + }); + // Cancel forwarding is independent of a writer blocked on disk I/O. + scope.spawn(|| { + while !shared.stop.load(Ordering::Acquire) { + if cancel.load(Ordering::Acquire) { + shared.stop.store(true, Ordering::Release); + shared.changed.notify_all(); + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + }); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + assemble(req, &shared, cancel, before_swap, progress) + })) + .unwrap_or_else(|_| Err("native assembly panicked".into())); + shared.stop.store(true, Ordering::Release); + { + let mut state = shared.state.lock().unwrap(); + state.done = true; + shared.changed.notify_all(); + } + result + }); + // No .part writer survives this point; preserve verified resident chunks for resume. + let resident = shared.state.lock().unwrap().schedule.oldest.clone(); + for index in resident { + let _ = fs::remove_file(super::chunk::part_path( + &req.cache.join(&req.chunks[index].guid), + )); + } + if result.is_ok() { + fs::remove_dir_all(&req.cache).map_err(|e| format!("clear cache: {e}"))?; + } + result +} + +fn await_chunk( + req: &Request, + shared: &Shared, + cancel: &AtomicBool, + c: usize, + fi: usize, + pi: usize, + stage_remaining: u64, +) -> Result<(), String> { + let mut state = shared.state.lock().unwrap(); + if cancel.load(Ordering::Acquire) { + return Err("cancelled".into()); + } + live_admit(req, &mut state, fi, pi, stage_remaining)?; + let fetch = state.schedule.fill(req); + state.pending.extend(fetch); + shared.changed.notify_all(); + while !state.ready[c] { + if cancel.load(Ordering::Acquire) { + return Err("cancelled".into()); + } + if let Some(error) = &state.error { + return Err(error.clone()); + } + if shared.stop.load(Ordering::Acquire) { + return Err("cancelled".into()); + } + state = shared + .changed + .wait_timeout(state, Duration::from_millis(100)) + .unwrap() + .0; + } + Ok(()) +} + +fn assemble( + req: &Request, + shared: &Shared, + cancel: &AtomicBool, + before_swap: &(dyn Fn() -> Result<(), String> + Sync), + progress: &(dyn Fn(u64, u64, u64, u64) + Sync), +) -> Result { + let mut done = 0; + let mut marked = false; + let total = req.total(); + for (fi, file) in req.files.iter().enumerate() { + Request::validate_staging(&req.install, &req.staging)?; + if cancel.load(Ordering::Acquire) { + return Err("cancelled".into()); + } + let output = req.install.join(&file.path); + fs::create_dir_all(output.parent().ok_or("missing output parent")?) + .map_err(|e| e.to_string())?; + let stage = req + .staging + .join(format!("native-{}-{fi}.stage", std::process::id())); + let result = (|| { + let stage_file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&stage) + .map_err(|e| e.to_string())?; + let mut out = BufWriter::with_capacity(64 * 1024, stage_file); + let mut sha = Sha1::new(); + let mut written = 0; + let file_total: u64 = file.parts.iter().map(|p| p.size).sum(); + for (pi, part) in file.parts.iter().enumerate() { + // A live free-space probe must include already buffered staging bytes. + out.flush().map_err(|e| e.to_string())?; + await_chunk( + req, + shared, + cancel, + part.chunk, + fi, + pi, + file_total - written, + )?; + written += super::assemble::copy_chunk_part( + &req.cache.join(&req.chunks[part.chunk].guid), + part.offset, + part.size, + &mut out, + &mut sha, + cancel, + )?; + let mut state = shared.state.lock().unwrap(); + for c in state.schedule.consume() { + if !state.ready[c] { + return Err("attempt to release in-flight chunk".into()); + } + fs::remove_file(req.cache.join(&req.chunks[c].guid)) + .map_err(|e| e.to_string())?; + state.ready[c] = false; + } + done += part.size; + progress(done, total, fi as u64, req.files.len() as u64); + } + out.flush().map_err(|e| e.to_string())?; + out.get_ref().sync_all().map_err(|e| e.to_string())?; + drop(out); + if written != file_total { + return Err("file size mismatch".into()); + } + if file + .sha + .is_some_and(|expected| sha.finalize().as_slice() != expected) + { + return Err(format!("SHA-1 mismatch for {}", file.path)); + } + if cancel.load(Ordering::Acquire) { + return Err("cancelled".into()); + } + if !marked { + before_swap()?; + marked = true; + } + if cancel.load(Ordering::Acquire) { + return Err("cancelled".into()); + } + // Android/Linux rename replaces atomically on the same output volume. Never delete first. + fs::rename(&stage, &output) + .map_err(|e| format!("atomic replacement {}: {e}", file.path))?; + progress(done, total, fi as u64 + 1, req.files.len() as u64); + Ok(()) + })(); + let _ = fs::remove_file(&stage); + result?; + } + Ok(done) +} + +#[cfg(test)] +mod tests { + use super::super::chunk::test_support::{build_chunk_body, sha1_of}; + use super::*; + use std::io::Read; + use std::net::TcpListener; + use std::sync::atomic::{AtomicU64, AtomicUsize}; + use std::sync::{Arc, Barrier}; + use std::thread; + use std::time::{Instant, SystemTime, UNIX_EPOCH}; + + static SEQUENCE: AtomicU64 = AtomicU64::new(0); + struct Fixture { + root: PathBuf, + req: Request, + } + impl Fixture { + fn new(count: usize, size: usize) -> Self { + let root = PathBuf::from("target").join(format!( + "epic-adaptive-{}-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(), + SEQUENCE.fetch_add(1, Ordering::Relaxed) + )); + let install = root.join("game"); + let cache = root.join("cache"); + let staging = install.join(".droiddeck-epic-staging"); + fs::create_dir_all(&staging).unwrap(); + fs::create_dir_all(&cache).unwrap(); + fs::write(staging.join("owner"), b"DroidDeck Epic staging v1\n").unwrap(); + let data = vec![42; size]; + let sha = sha1_of(&data); + let chunks: Vec<_> = (0..count) + .map(|c| Chunk { + guid: format!("{c:032X}"), + path: format!("chunks/{c}"), + size: size as u64, + wire: size as u64 + 41, + sha: Some(sha), + cached: None, + exact: true, + }) + .collect(); + let files = (0..count) + .map(|c| Output { + path: format!("file-{c}.bin"), + sha: Some(sha), + old: 0, + parts: vec![Part { + chunk: c, + offset: 0, + size: size as u64, + }], + }) + .collect(); + Self { + root, + req: Request { + install, + cache, + staging, + chunks, + files, + same_volume: false, + free_game: Some(u64::MAX), + free_cache: Some(u64::MAX), + hosts: vec!["http://127.0.0.1:1".into()], + max_workers: 8, + }, + } + } + } + impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.root); + } + } + + struct Server { + address: String, + requests: Arc, + active: Arc, + stop: Arc, + worker: Option>, + } + impl Server { + fn start(data: Vec, barrier_count: usize, stalled: bool) -> Self { + Self::start_mode(data, barrier_count, stalled.then_some(usize::MAX)) + } + + fn start_mode(data: Vec, barrier_count: usize, stall_chunk: Option) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = format!("http://{}", listener.local_addr().unwrap()); + listener.set_nonblocking(true).unwrap(); + let requests = Arc::new(AtomicUsize::new(0)); + let active = Arc::new(AtomicUsize::new(0)); + let stop = Arc::new(AtomicBool::new(false)); + let requests2 = requests.clone(); + let active2 = active.clone(); + let stop2 = stop.clone(); + let worker = thread::spawn(move || { + let body = Arc::new(build_chunk_body(&data, false, 0, None)); + let barrier = Arc::new(Barrier::new(barrier_count.max(1))); + let mut handlers = Vec::new(); + while !stop2.load(Ordering::Acquire) { + let (mut socket, _) = match listener.accept() { + Ok(s) => s, + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(5)); + continue; + } + Err(e) => panic!("{e}"), + }; + let body = body.clone(); + let requests = requests2.clone(); + let active = active2.clone(); + let stop = stop2.clone(); + let barrier = barrier.clone(); + handlers.push(thread::spawn(move || { + active.fetch_add(1, Ordering::Relaxed); + let _ = socket.set_read_timeout(Some(Duration::from_secs(5))); + let mut request = Vec::new(); + let mut byte = [0]; + while socket.read(&mut byte).unwrap_or(0) > 0 { + request.push(byte[0]); + if request.ends_with(b"\r\n\r\n") { + break; + } + } + let n = requests.fetch_add(1, Ordering::Relaxed); + if n < barrier_count { + barrier.wait(); + } + let _ = write!( + socket, + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let stalled = stall_chunk.is_some_and(|c| { + c == usize::MAX + || String::from_utf8_lossy(&request) + .contains(&format!("/chunks/{c} ")) + }); + if stalled { + // Cancellation must close the native client's connection without data. + let _ = socket.set_read_timeout(Some(Duration::from_millis(100))); + loop { + if stop.load(Ordering::Acquire) { + break; + } + match socket.read(&mut byte) { + Ok(0) => break, + Ok(_) => {} + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => {} + Err(_) => break, + } + } + } else { + let _ = socket.write_all(&body); + } + active.fetch_sub(1, Ordering::Relaxed); + })); + } + for handler in handlers { + handler.join().unwrap(); + } + }); + Self { + address, + requests, + active, + stop, + worker: Some(worker), + } + } + } + impl Drop for Server { + fn drop(&mut self) { + self.stop.store(true, Ordering::Release); + if let Some(worker) = self.worker.take() { + worker.join().unwrap(); + } + } + } + + #[test] + fn situation_table_selects_large_tight_minimal_and_rejects_no_fit() { + let mut f = Fixture::new(300, MIB as usize); + assert_eq!(choose_budget(&f.req).unwrap().lookahead, 128); + f.req.free_cache = Some(8 * MIB); + assert_eq!(choose_budget(&f.req).unwrap().lookahead, 8); + f.req.free_cache = Some(MIB); + assert_eq!(choose_budget(&f.req).unwrap().lookahead, 1); + f.req.free_cache = Some(MIB - 1); + assert!(choose_budget(&f.req) + .unwrap_err() + .contains("Not enough free space")); + f.req.free_cache = None; + assert_eq!(choose_budget(&f.req).unwrap().lookahead, 8); + f.req.free_game = Some(0); + assert!(choose_budget(&f.req).is_err()); + } + + #[test] + fn planner_counts_all_incoming_old_outputs_and_existing_cache_release() { + let mut f = Fixture::new(20, 100); + f.req.same_volume = true; + for file in &mut f.req.files { + file.old = 100; + } + assert_eq!(peaks(&f.req, 8, 0), (900, 800)); + for chunk in &mut f.req.chunks { + chunk.cached = Some(100); + } + assert_eq!(peaks(&f.req, 8, 0), (100, 0)); + f.req.same_volume = false; + assert_eq!(peaks(&f.req, 8, 0), (100, 0)); + } + + #[test] + fn shared_references_deduplicate_and_distant_chunks_refetch_under_budget() { + let mut f = Fixture::new(12, (40 * MIB) as usize); + f.req.files.extend(f.req.files.clone()); + let mut schedule = Schedule::new(&f.req, 8, RETAINED); + assert_eq!(schedule.fill(&f.req), (0..8).collect::>()); + assert!(schedule.fill(&f.req).is_empty()); + assert!(schedule.consume().is_empty()); + assert_eq!(schedule.fill(&f.req), vec![8]); + assert_eq!(schedule.consume(), vec![0]); + let mut refetched = false; + for _ in 2..24 { + refetched |= schedule.fill(&f.req).contains(&0); + assert!(schedule.bytes() <= RETAINED + 8 * 40 * MIB); + let removed = schedule.consume(); + assert!(removed.iter().all(|c| !schedule.window().contains(c))); + } + assert!(refetched); + assert_eq!(schedule.bytes(), 0); + } + + #[test] + fn native_fetch_overlaps_across_files_and_never_fetches_outside_window() { + let mut f = Fixture::new(30, 4096); + let server = Server::start(vec![42; 4096], 8, false); + f.req.hosts = vec![server.address.clone()]; + f.req.free_cache = Some(8 * 4096); + let budget = choose_budget(&f.req).unwrap(); + assert_eq!(budget.lookahead, 8); + let cancel = AtomicBool::new(false); + let swapped = AtomicUsize::new(0); + let progress = |_, _, files, _| { + if files == 0 { + cancel.store(true, Ordering::Release); + } + }; + let result = run( + &f.req, + budget, + &cancel, + &|| { + swapped.fetch_add(1, Ordering::Relaxed); + Ok(()) + }, + &progress, + &|_| {}, + ); + assert!(result.is_err()); + assert_eq!(server.requests.load(Ordering::Relaxed), 8); + assert_eq!(swapped.load(Ordering::Relaxed), 0); + assert_eq!(fs::read_dir(&f.req.staging).unwrap().count(), 1); + assert!(fs::read_dir(&f.req.cache).unwrap().all(|e| !e + .unwrap() + .path() + .to_string_lossy() + .ends_with(".part"))); + } + + #[test] + fn native_success_stages_verifies_swaps_and_reclaims_both_cache_layouts() { + for same_volume in [false, true] { + let mut f = Fixture::new(20, 4096); + let server = Server::start(vec![42; 4096], 8, false); + f.req.hosts = vec![server.address.clone()]; + f.req.same_volume = same_volume; + if same_volume { + f.req.cache = f.req.install.join(".chunks"); + fs::create_dir_all(&f.req.cache).unwrap(); + } + for file in &mut f.req.files { + fs::write(f.req.install.join(&file.path), b"old").unwrap(); + file.old = 3; + } + let swaps = AtomicUsize::new(0); + let before = || { + assert_eq!(fs::read(f.req.install.join("file-0.bin")).unwrap(), b"old"); + swaps.fetch_add(1, Ordering::Relaxed); + Ok(()) + }; + assert_eq!( + run( + &f.req, + choose_budget(&f.req).unwrap(), + &AtomicBool::new(false), + &before, + &|_, _, _, _| {}, + &|_| {} + ) + .unwrap(), + 20 * 4096 + ); + assert_eq!(swaps.load(Ordering::Relaxed), 1); + assert!(!f.req.cache.exists()); + for file in &f.req.files { + assert_eq!( + fs::read(f.req.install.join(&file.path)).unwrap(), + vec![42; 4096] + ); + } + assert_eq!(fs::read_dir(&f.req.staging).unwrap().count(), 1); + } + } + + #[test] + fn native_file_hash_failure_and_updating_failure_keep_old_output() { + for updating_failure in [false, true] { + let mut f = Fixture::new(1, 4096); + let server = Server::start(vec![42; 4096], 1, false); + f.req.hosts = vec![server.address.clone()]; + fs::write(f.req.install.join("file-0.bin"), b"old").unwrap(); + if !updating_failure { + f.req.files[0].sha = Some([1; 20]); + } + let before = || Err("updating record failed".into()); + let result = run( + &f.req, + choose_budget(&f.req).unwrap(), + &AtomicBool::new(false), + &before, + &|_, _, _, _| {}, + &|_| {}, + ); + assert!(result.is_err()); + assert_eq!(fs::read(f.req.install.join("file-0.bin")).unwrap(), b"old"); + assert_eq!(fs::read_dir(&f.req.staging).unwrap().count(), 1); + } + } + + #[test] + fn native_cancellation_closes_stalled_parallel_requests_and_joins_before_return() { + let mut f = Fixture::new(30, 4096); + let server = Server::start(vec![42; 4096], 8, true); + f.req.hosts = vec![server.address.clone()]; + f.req.free_cache = Some(8 * 4096); + let cancel = AtomicBool::new(false); + thread::scope(|scope| { + scope.spawn(|| { + let until = Instant::now() + Duration::from_secs(5); + while server.requests.load(Ordering::Relaxed) < 8 && Instant::now() < until { + thread::sleep(Duration::from_millis(10)); + } + cancel.store(true, Ordering::Release); + }); + let started = Instant::now(); + assert!(run( + &f.req, + choose_budget(&f.req).unwrap(), + &cancel, + &|| Ok(()), + &|_, _, _, _| {}, + &|_| {} + ) + .is_err()); + assert!(started.elapsed() < Duration::from_secs(7)); + }); + let until = Instant::now() + Duration::from_secs(2); + while server.active.load(Ordering::Relaxed) > 0 && Instant::now() < until { + thread::sleep(Duration::from_millis(10)); + } + assert_eq!(server.active.load(Ordering::Relaxed), 0); + assert_eq!(server.requests.load(Ordering::Relaxed), 8); + assert_eq!(fs::read_dir(&f.req.staging).unwrap().count(), 1); + assert_eq!(fs::read_dir(&f.req.cache).unwrap().count(), 0); + } + + #[test] + fn native_bounded_decode_never_publishes_oversized_or_truncated_chunks() { + let f = Fixture::new(1, 10); + let path = f.req.cache.join("chunk"); + let body = build_chunk_body(&[42; 11], true, 0, None); + assert!( + super::super::chunk::write_verified_chunk_bounded(&body, None, &path, 10, true) + .is_err() + ); + assert!(!path.exists()); + assert!(!super::super::chunk::part_path(&path).exists()); + let body = build_chunk_body(&[42; 10], false, 0, Some(11)); + assert!( + super::super::chunk::write_verified_chunk_bounded(&body, None, &path, 10, true) + .is_err() + ); + let data = vec![42; 512 * 1024]; + let mut compressed = build_chunk_body(&data, true, 0, None); + assert_eq!( + super::super::chunk::write_verified_chunk_bounded( + &compressed, + Some(&sha1_of(&data)), + &path, + data.len() as u64, + true + ) + .unwrap(), + data.len() as u64 + ); + fs::remove_file(&path).unwrap(); + compressed.pop(); + // A matching shorter payload header leaves the missing zlib stream end to detect failure. + let short = (compressed.len() - 41) as i32; + compressed[12..16].copy_from_slice(&short.to_le_bytes()); + assert!(super::super::chunk::write_verified_chunk_bounded( + &compressed, + None, + &path, + data.len() as u64, + false + ) + .is_err()); + assert!(!path.exists()); + } + + #[test] + fn live_admission_shrinks_unsent_work_but_never_withdraws_inflight_reservations() { + let f = Fixture::new(30, 4096); + let mut state = State { + schedule: Schedule::new(&f.req, 128, RETAINED), + ready: vec![false; 30], + writing: vec![false; 30], + disk_sizes: vec![0; 30], + pending: vec![], + error: None, + done: false, + }; + admit_with_probes( + &f.req, + &mut state, + 0, + 0, + 4096, + 0, + Some(u64::MAX), + Some(8 * 4096), + ) + .unwrap(); + assert_eq!(state.schedule.lookahead, 8); + assert_eq!(state.schedule.retained, 0); + assert_eq!(state.schedule.fill(&f.req).len(), 8); + assert!(admit_with_probes( + &f.req, + &mut state, + 0, + 0, + 4096, + 0, + Some(u64::MAX), + Some(4096) + ) + .is_err()); + assert_eq!(state.schedule.lookahead, 8); + assert!(state + .schedule + .oldest + .iter() + .all(|&c| state.schedule.resident[c].is_some())); + } + + #[test] + fn snapshot_schema_rejects_unowned_staging_bad_paths_and_bad_version() { + let f = Fixture::new(1, 10); + let mut snapshot = serde_json::json!({ + "schema_version": 1, + "install_dir": f.req.install, "cache_dir": f.req.cache, "staging_dir": f.req.staging, + "same_volume": false, "free_game": -1, "free_cache": 0, "max_workers": 32, + "hosts": ["http://127.0.0.1"], + "chunks": [{ "guid": "00000000000000000000000000000000", "path": "chunks/00/chunk", + "size": 10, "wire_size": 51, "sha1": "", "cached": -1, "exact_size": true }], + "files": [{ "path": "legitimate.part", "old_size": 0, "sha1": "", + "parts": [{ "guid": "00000000000000000000000000000000", "offset": 0, "size": 10 }] }] + }); + let parsed = Request::parse(&snapshot.to_string()).unwrap(); + assert_eq!( + parsed.free_game, + if parsed.same_volume { Some(0) } else { None } + ); + assert_eq!(parsed.free_cache, Some(0)); + assert_eq!(parsed.chunks[0].cached, None); + assert!(choose_budget(&parsed).is_err()); + snapshot["schema_version"] = 2.into(); + assert!(Request::parse(&snapshot.to_string()).is_err()); + snapshot["schema_version"] = 1.into(); + snapshot["files"][0]["path"] = "../outside".into(); + assert!(Request::parse(&snapshot.to_string()).is_err()); + snapshot["files"][0]["path"] = ".droiddeck-epic-staging/owner".into(); + assert!(Request::parse(&snapshot.to_string()).is_err()); + snapshot["files"][0]["path"] = "legitimate.part".into(); + fs::remove_file(f.req.staging.join("owner")).unwrap(); + assert!(Request::parse(&snapshot.to_string()).is_err()); + } + + #[test] + fn native_shared_chunk_survives_cancelled_file_boundary_and_is_reused_on_resume() { + let mut f = Fixture::new(3, 4096); + let server = Server::start(vec![42; 4096], 1, false); + f.req.hosts = vec![server.address.clone()]; + for file in &mut f.req.files { + file.parts[0].chunk = 0; + } + let cancel = AtomicBool::new(false); + let before = AtomicUsize::new(0); + let progress = |_, _, files, _| { + if files == 1 { + cancel.store(true, Ordering::Release); + } + }; + assert!(run( + &f.req, + choose_budget(&f.req).unwrap(), + &cancel, + &|| { + before.fetch_add(1, Ordering::Relaxed); + Ok(()) + }, + &progress, + &|_| {} + ) + .is_err()); + assert_eq!(before.load(Ordering::Relaxed), 1); + assert_eq!(server.requests.load(Ordering::Relaxed), 1); + assert_eq!( + fs::read(f.req.install.join("file-0.bin")).unwrap(), + vec![42; 4096] + ); + assert!(!f.req.install.join("file-1.bin").exists()); + assert!(f.req.cache.join(&f.req.chunks[0].guid).is_file()); + f.req.files.remove(0); + f.req.chunks[0].cached = Some(4096); + assert_eq!( + run( + &f.req, + choose_budget(&f.req).unwrap(), + &AtomicBool::new(false), + &|| Ok(()), + &|_, _, _, _| {}, + &|_| {} + ) + .unwrap(), + 8192 + ); + assert_eq!(server.requests.load(Ordering::Relaxed), 1); + assert!(!f.req.cache.exists()); + } + + #[test] + fn native_later_hash_failure_preserves_verified_swap_and_unreplaced_old_file() { + let mut f = Fixture::new(2, 4096); + let server = Server::start(vec![42; 4096], 2, false); + f.req.hosts = vec![server.address.clone()]; + for file in &mut f.req.files { + fs::write(f.req.install.join(&file.path), b"old").unwrap(); + file.old = 3; + } + f.req.files[1].sha = Some([1; 20]); + let marked = AtomicUsize::new(0); + let error = run( + &f.req, + choose_budget(&f.req).unwrap(), + &AtomicBool::new(false), + &|| { + marked.fetch_add(1, Ordering::Relaxed); + Ok(()) + }, + &|_, _, _, _| {}, + &|_| {}, + ) + .unwrap_err(); + assert!(error.contains("SHA-1 mismatch")); + assert_eq!(marked.load(Ordering::Relaxed), 1); + assert_eq!( + fs::read(f.req.install.join("file-0.bin")).unwrap(), + vec![42; 4096] + ); + assert_eq!(fs::read(f.req.install.join("file-1.bin")).unwrap(), b"old"); + assert_eq!(fs::read_dir(&f.req.staging).unwrap().count(), 1); + } + + #[test] + fn actual_native_file_and_cache_growth_never_exceeds_the_shared_planned_peak() { + let mut f = Fixture::new(20, 4096); + let server = Server::start(vec![42; 4096], 8, false); + f.req.hosts = vec![server.address.clone()]; + f.req.same_volume = true; + f.req.free_game = Some(21 * 4096); + for file in &mut f.req.files { + file.old = 4096; + fs::write(f.req.install.join(&file.path), vec![1; 4096]).unwrap(); + } + let budget = choose_budget(&f.req).unwrap(); + let observed = AtomicU64::new(0); + let progress = |_, _, _, _| { + let game: u64 = f + .req + .files + .iter() + .map(|file| fs::metadata(f.req.install.join(&file.path)).unwrap().len()) + .sum(); + let stage: u64 = fs::read_dir(&f.req.staging) + .unwrap() + .filter_map(Result::ok) + .filter(|e| e.file_name() != "owner") + .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0)) + .sum(); + let cache: u64 = fs::read_dir(&f.req.cache) + .unwrap() + .filter_map(Result::ok) + .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0)) + .sum(); + let additional = (game + stage + cache).saturating_sub(20 * 4096); + observed.fetch_max(additional, Ordering::Relaxed); + assert!( + additional <= budget.game_peak, + "{additional} > planned {}", + budget.game_peak + ); + }; + assert_eq!( + run( + &f.req, + budget, + &AtomicBool::new(false), + &|| Ok(()), + &progress, + &|_| {} + ) + .unwrap(), + 20 * 4096 + ); + assert!(observed.load(Ordering::Relaxed) > 8 * 4096); + } + + #[test] + fn native_assembly_swaps_a_verified_file_while_a_future_chunk_is_still_fetching() { + let mut f = Fixture::new(20, 4096); + let server = Server::start_mode(vec![42; 4096], 8, Some(7)); + f.req.hosts = vec![server.address.clone()]; + f.req.free_cache = Some(8 * 4096); + let cancel = AtomicBool::new(false); + let overlap = AtomicBool::new(false); + let progress = |_, _, files, _| { + if files == 1 { + assert!(server.active.load(Ordering::Relaxed) > 0); + assert_eq!( + fs::read(f.req.install.join("file-0.bin")).unwrap(), + vec![42; 4096] + ); + overlap.store(true, Ordering::Release); + cancel.store(true, Ordering::Release); + } + }; + assert!(run( + &f.req, + choose_budget(&f.req).unwrap(), + &cancel, + &|| Ok(()), + &progress, + &|_| {} + ) + .is_err()); + assert!(overlap.load(Ordering::Acquire)); + assert_eq!(server.requests.load(Ordering::Relaxed), 8); + assert_eq!(fs::read_dir(&f.req.staging).unwrap().count(), 1); + } + + #[cfg(unix)] + #[test] + fn staging_validation_rejects_symlink_namespace_owner_and_entries() { + use std::os::unix::fs::symlink; + let f = Fixture::new(1, 4096); + let outside = f.root.join("outside"); + fs::create_dir(&outside).unwrap(); + fs::write(outside.join("owner"), b"DroidDeck Epic staging v1\n").unwrap(); + fs::write(outside.join("keep.stage"), b"keep").unwrap(); + fs::remove_dir_all(&f.req.staging).unwrap(); + symlink(fs::canonicalize(&outside).unwrap(), &f.req.staging).unwrap(); + assert!(Request::validate_staging(&f.req.install, &f.req.staging).is_err()); + assert!(run( + &f.req, + choose_budget(&f.req).unwrap(), + &AtomicBool::new(false), + &|| Ok(()), + &|_, _, _, _| {}, + &|_| {} + ) + .is_err()); + assert_eq!(fs::read(outside.join("keep.stage")).unwrap(), b"keep"); + fs::remove_file(&f.req.staging).unwrap(); + fs::create_dir(&f.req.staging).unwrap(); + symlink( + fs::canonicalize(outside.join("owner")).unwrap(), + f.req.staging.join("owner"), + ) + .unwrap(); + assert!(Request::validate_staging(&f.req.install, &f.req.staging).is_err()); + fs::remove_file(f.req.staging.join("owner")).unwrap(); + fs::write(f.req.staging.join("owner"), b"DroidDeck Epic staging v1\n").unwrap(); + symlink( + fs::canonicalize(outside.join("keep.stage")).unwrap(), + f.req.staging.join("linked.stage"), + ) + .unwrap(); + assert!(Request::validate_staging(&f.req.install, &f.req.staging).is_err()); + assert_eq!(fs::read(outside.join("keep.stage")).unwrap(), b"keep"); + fs::remove_file(f.req.staging.join("linked.stage")).unwrap(); + } +} diff --git a/app/src/main/rust/stores/src/epic/assemble.rs b/app/src/main/rust/stores/src/epic/assemble.rs index 580cb8c24..46edf3107 100644 --- a/app/src/main/rust/stores/src/epic/assemble.rs +++ b/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 { + 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, diff --git a/app/src/main/rust/stores/src/epic/chunk.rs b/app/src/main/rust/stores/src/epic/chunk.rs index 22ebb864d..9635b5983 100644 --- a/app/src/main/rust/stores/src/epic/chunk.rs +++ b/app/src/main/rust/stores/src/epic/chunk.rs @@ -93,10 +93,14 @@ struct HashingWriter { inner: W, sha: Option, written: u64, + limit: u64, } impl Write for HashingWriter { fn write(&mut self, buf: &[u8]) -> io::Result { + 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 { + 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 { 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[..] { diff --git a/app/src/main/rust/stores/src/epic/jni.rs b/app/src/main/rust/stores/src/epic/jni.rs index 937d6e42a..c43fe4b1f 100644 --- a/app/src/main/rust/stores/src/epic/jni.rs +++ b/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, + worker: Option>, } 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 } diff --git a/app/src/test/java/com/droiddeck/launcher/stores/epic/EpicBoundedInstallTest.kt b/app/src/test/java/com/droiddeck/launcher/stores/epic/EpicBoundedInstallTest.kt index 12e90e94b..cfb701191 100644 --- a/app/src/test/java/com/droiddeck/launcher/stores/epic/EpicBoundedInstallTest.kt +++ b/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() + 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() + 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) + } + } + } } diff --git a/app/src/test/java/com/droiddeck/launcher/stores/epic/EpicJsonManifestTest.kt b/app/src/test/java/com/droiddeck/launcher/stores/epic/EpicJsonManifestTest.kt index 8c34c125d..30722f9e5 100644 --- a/app/src/test/java/com/droiddeck/launcher/stores/epic/EpicJsonManifestTest.kt +++ b/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(), schedule.fill()) + assertEquals(8 * size, schedule.bytes()) + assertEquals(emptyList(), 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()}") diff --git a/docs/development/stores.md b/docs/development/stores.md index 4ef41eaad..8eb7233a3 100644 --- a/docs/development/stores.md +++ b/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 `.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 `.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 `/.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