Pace Epic updates with a replacement-overlap budget

A staged file that replaces an existing one sits beside its old copy
until its swap, so many large replaced files open at once could run a
nearly full device out of space mid-update. The stream engine now admits
files in order and admits a replaced file only while the planned sizes of
open replaced files fit a budget (always at least one); the fetch core
dispatches a chunk only once its first file is admitted. Fresh files are
never held back. The manager sizes the budget from the game volume's free
space after the net growth and the cache, floored at the largest replaced
file, and hands over to the lookahead engine when that does not fit.
A successful install also removes the empty owned staging namespace.

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

@@ -81,12 +81,6 @@ public final class EpicDownloadManager {
     static final long ADAPTIVE_CACHE_MIN_SCRATCH_BYTES = MAX_RETAINED_CHUNK_CACHE_BYTES;
     /** Chunks being decoded, written or released at any moment, beyond the budget itself. */
     static final long ADAPTIVE_TRANSIENT_BYTES = 64L * 1024 * 1024;
-    /**
-     * Files the fetch window has open beyond the sequential plan: the parallel writer can finish
-     * files a little out of order, so replaced files near the front of the window may not have
-     * released their old copies yet.
-     */
-    static final long ADAPTIVE_OPEN_WINDOW_BYTES = 256L * 1024 * 1024;
 
     static final String MODE_ADAPTIVE = "adaptive";
     static final String MODE_LOOKAHEAD = "lookahead";
@@ -231,6 +225,15 @@ public final class EpicDownloadManager {
     static Result installManifest(Manifest manifest, String installDirPath, String chunkCacheDirPath,
                                   List<String> installTags, AtomicBoolean cancel, Callback cb,
                                   NativeOptions nativeOptions) throws InstallException {
+        Result result = installSelected(manifest, installDirPath, chunkCacheDirPath, installTags, cancel, cb, nativeOptions);
+        // A finished install leaves no staging; the owned namespace is recreated by the next run.
+        if (result != null) releaseOwnedStaging(new File(installDirPath));
+        return result;
+    }
+
+    private static Result installSelected(Manifest manifest, String installDirPath, String chunkCacheDirPath,
+                                          List<String> installTags, AtomicBoolean cancel, Callback cb,
+                                          NativeOptions nativeOptions) throws InstallException {
         try {
             List<CdnUrl> cdnUrls = manifest.cdnUrls;
             File installDir = new File(installDirPath);
@@ -470,6 +473,11 @@ public final class EpicDownloadManager {
     static final class AdaptiveSpacePlan {
         boolean cacheBesideGame;
         long cacheBudget;
+        /**
+         * Most planned bytes of replaced files the engine may stage at once (each sits beside its
+         * old copy until its swap); -1 = nothing is replaced, no limit.
+         */
+        long overlapBudget = -1;
         long gameNeed;
         long scratchNeed;
         String error;
@@ -484,30 +492,57 @@ public final class EpicDownloadManager {
     }
 
     /**
-     * Space the game volume needs beyond the sequential staged plan: none for a fresh install (a
-     * staged file becomes the file itself); for replaced files, the largest old copy that may not
-     * be released yet plus what the fetch window spans, never more than all old copies together.
+     * The game volume's net growth at its worst point when files are swapped in order and each
+     * replaced file's old copy is released at its swap: the running Σ(new − old) after each file.
+     * What a staged file costs while it sits beside its old copy is the overlap budget's part.
      */
-    static long adaptiveOpenStagingAllowance(Map<String, Long> oldFileBytes) {
-        long total = 0, largest = 0;
-        for (long bytes : oldFileBytes.values()) { total += bytes; largest = Math.max(largest, bytes); }
-        return Math.min(total, largest + ADAPTIVE_OPEN_WINDOW_BYTES);
+    static long settledNetPeak(List<FileInfo> files, Map<String, Long> oldFileBytes) {
+        long running = 0, peak = 0;
+        for (FileInfo file : files) {
+            Long old = oldFileBytes.get(file.filename);
+            running += file.fileSize() - (old == null ? 0 : old);
+            peak = Math.max(peak, running);
+        }
+        return peak;
     }
 
-    static AdaptiveSpacePlan planAdaptiveSpace(long stagedPeak, long openAllowance, long freeGame, boolean scratchRequested, long freeScratch) {
+    /** The largest pending file whose destination has data (the engine's smallest overlap budget). */
+    static long largestReplacedFile(List<FileInfo> files, Map<String, Long> oldFileBytes) {
+        long largest = 0;
+        for (FileInfo file : files) {
+            Long old = oldFileBytes.get(file.filename);
+            if (old != null && old > 0) largest = Math.max(largest, file.fileSize());
+        }
+        return largest;
+    }
+
+    /**
+     * The cache placement and budget as before; for an update, the replacement-overlap budget is
+     * what the game volume has left after the net growth and the cache, never below the largest
+     * replaced file (one at a time is the minimum that still progresses). The check is the
+     * lookahead planner's: the need must not exceed the free reading, no extra margin.
+     */
+    static AdaptiveSpacePlan planAdaptiveSpace(long netPeak, long largestReplaced, long freeGame, boolean scratchRequested, long freeScratch) {
         AdaptiveSpacePlan plan = new AdaptiveSpacePlan();
-        long files = stagedPeak + openAllowance;
+        long minimum = netPeak + largestReplaced;
         long scratchBudget = scratchRequested ? adaptiveCacheBudget(freeScratch, 0) : 0;
         plan.cacheBesideGame = !scratchRequested || (freeScratch >= 0 && scratchBudget < ADAPTIVE_CACHE_MIN_SCRATCH_BYTES);
+        long cacheOnGame;
         if (plan.cacheBesideGame) {
-            plan.cacheBudget = adaptiveCacheBudget(freeGame, files);
-            plan.gameNeed = files + plan.cacheBudget + ADAPTIVE_TRANSIENT_BYTES;
+            plan.cacheBudget = adaptiveCacheBudget(freeGame, minimum);
+            cacheOnGame = plan.cacheBudget + ADAPTIVE_TRANSIENT_BYTES;
             plan.scratchNeed = 0;
         } else {
             plan.cacheBudget = scratchBudget;
-            plan.gameNeed = files;
+            cacheOnGame = 0;
             plan.scratchNeed = plan.cacheBudget + ADAPTIVE_TRANSIENT_BYTES;
         }
+        long overlap = 0;
+        if (largestReplaced > 0) {
+            overlap = freeGame < 0 ? largestReplaced : Math.max(largestReplaced, freeGame - netPeak - cacheOnGame);
+            plan.overlapBudget = overlap;
+        }
+        plan.gameNeed = netPeak + overlap + cacheOnGame;
         if (StoreInstallRoot.insufficientSpace(plan.gameNeed, freeGame)) {
             plan.error = "Not enough free space on the game's storage: need " + fmt(plan.gameNeed)
                     + (plan.cacheBesideGame && scratchRequested ? " (download cache moved beside the game: internal storage is full)" : "")
@@ -531,12 +566,12 @@ public final class EpicDownloadManager {
         if (manifest.rawBytes == null || pendingIdx == null) {
             throw new InstallException("Native Epic install failed: the manifest is not available to the native engine");
         }
-        long stagedPeak = estimateStagedFilePeak(pending, oldFileBytes);
-        long openAllowance = adaptiveOpenStagingAllowance(oldFileBytes);
+        long netPeak = settledNetPeak(pending, oldFileBytes);
+        long largestReplaced = largestReplacedFile(pending, oldFileBytes);
         long freeGame = StoreInstallRoot.availableBytes(installDir);
         boolean scratchRequested = !cachePath.isEmpty();
         long freeScratch = scratchRequested ? StoreInstallRoot.availableBytes(chunkCacheDir) : -1;
-        AdaptiveSpacePlan space = planAdaptiveSpace(stagedPeak, openAllowance, freeGame, scratchRequested, freeScratch);
+        AdaptiveSpacePlan space = planAdaptiveSpace(netPeak, largestReplaced, freeGame, scratchRequested, freeScratch);
         if (space.error != null) {
             cb.onLog("epic: adaptive space plan does not fit (" + space.error + "); mode=lookahead");
             return NativeAttempt.FALLBACK;
@@ -549,6 +584,7 @@ public final class EpicDownloadManager {
         if (cancel.get()) return NativeAttempt.done(null);
         cb.onLog("epic: mode=adaptive tier=" + options.tierId + " workers=" + options.maxWorkers + " process_workers=" + options.processWorkers
                 + " cache=" + (engineCache.isEmpty() ? "beside the game" : "scratch") + " budget=" + fmt(space.cacheBudget)
+                + " overlap_budget=" + (space.overlapBudget < 0 ? "none" : fmt(space.overlapBudget))
                 + " game need=" + fmt(space.gameNeed) + " of " + fmt(freeGame) + (space.cacheBesideGame ? "" : ", scratch need=" + fmt(space.scratchNeed) + " of " + fmt(freeScratch)));
         cb.onSizes(pendingBytes, planned);
         cb.onBytes(0, pendingBytes, 0);
@@ -558,7 +594,7 @@ public final class EpicDownloadManager {
         final long started = System.nanoTime();
         final long[] state = new long[] {0, 0, System.currentTimeMillis(), 0, 0}; // bytes, files, speed ms, speed bytes, speed bps
         EpicNative.Result res = EpicNative.stream(manifest.rawBytes, installDir.getPath(), engineCache, stagingDir.getCanonicalPath(), cdnPrefixes(manifest.cdnUrls), pendingIdx, pendingBytes,
-                space.cacheBudget, "", options.maxWorkers, options.processWorkers, cancel, new EpicNative.StreamListener() {
+                space.cacheBudget, space.overlapBudget, "", options.maxWorkers, options.processWorkers, cancel, new EpicNative.StreamListener() {
                     @Override public void onStreamProgress(long bytesWritten, long bytesTotal, int filesDone, int filesTotal, long filesBytes) {
                         synchronized (lock) {
                             if (bytesWritten < state[0] && filesDone <= state[1]) return; // a late, older sample
@@ -734,6 +770,28 @@ public final class EpicDownloadManager {
         return staging;
     }
 
+    /**
+     * Remove the owned staging namespace when it holds nothing but its own marker. Same no-follow
+     * and canonical-path checks as {@link #recoverOwnedStaging}; anything unexpected is left alone.
+     */
+    static void releaseOwnedStaging(File installDir) {
+        try {
+            File staging = new File(installDir, STAGING_DIRECTORY);
+            if (!Files.isDirectory(staging.toPath(), LinkOption.NOFOLLOW_LINKS)) return;
+            Path expected = installDir.getCanonicalFile().toPath().resolve(STAGING_DIRECTORY);
+            if (!staging.getCanonicalFile().toPath().equals(expected)) return;
+            String[] names = staging.list();
+            if (names == null || names.length != 1 || !names[0].equals("owner")) return;
+            File owner = new File(staging, "owner");
+            if (!Files.isRegularFile(owner.toPath(), LinkOption.NOFOLLOW_LINKS)
+                    || owner.length() != STAGING_OWNER.length()
+                    || !STAGING_OWNER.equals(new String(Files.readAllBytes(owner.toPath()), StandardCharsets.UTF_8))) return;
+            if (owner.delete()) staging.delete();
+        } catch (IOException e) {
+            Log.w(TAG, "staging namespace kept: " + e.getMessage());
+        }
+    }
+
     private static JSONObject nativeRequest(Manifest manifest, List<FileInfo> files,
             Map<String, ChunkInfo> chunks, Map<String, Long> cached, Map<String, Long> old,
             File install, File cache, File staging, boolean sameVolume) throws Exception {

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

@@ -90,13 +90,14 @@ object EpicNative {
 
     /**
      * The adaptive install of the files [pendingFileIdx] (indices into the manifest's file list),
-     * with at most [cacheBudget] bytes of shared chunks cached, staged in [stagingDir] (the
+     * with at most [cacheBudget] bytes of shared chunks cached and at most [overlapBudget] planned
+     * bytes of replaced files staged at once (`-1` = no limit), staged in [stagingDir] (the
      * manager's owned staging directory). `started == false` means nothing was fetched or written.
      */
     @JvmStatic
     fun stream(
         manifest: ByteArray, installDir: String, chunkCacheDir: String, stagingDir: String, cdnPrefixes: Array<String>, pendingFileIdx: IntArray, expectedBytes: Long,
-        cacheBudget: Long, caBundlePath: String, maxWorkers: Int, processWorkers: Int, cancel: AtomicBoolean?, listener: StreamListener,
+        cacheBudget: Long, overlapBudget: Long, caBundlePath: String, maxWorkers: Int, processWorkers: Int, cancel: AtomicBoolean?, listener: StreamListener,
     ): Result {
         if (!StoresNative.ensureLoaded()) return Result(false, false, false, "engine not built", 0L, 0, 0)
         val latch = CountDownLatch(1)
@@ -117,7 +118,7 @@ object EpicNative {
             }
         }
         val handle: Long = try {
-            nativeStream(manifest, installDir, chunkCacheDir, stagingDir, cdnPrefixes, pendingFileIdx, expectedBytes, cacheBudget, caBundlePath, maxWorkers, processWorkers, inner)
+            nativeStream(manifest, installDir, chunkCacheDir, stagingDir, cdnPrefixes, pendingFileIdx, expectedBytes, cacheBudget, overlapBudget, caBundlePath, maxWorkers, processWorkers, inner)
         } catch (t: Throwable) {
             Log.w(TAG, "engine unavailable: ${t.javaClass.simpleName}: ${t.message}")
             return Result(false, false, false, "nativeStream: ${t.javaClass.simpleName}: ${t.message}", 0L, 0, 0)
@@ -205,7 +206,7 @@ object EpicNative {
     @JvmStatic
     private external fun nativeStream(
         manifest: ByteArray, installDir: String, chunkCacheDir: String, stagingDir: String, cdnPrefixes: Array<String>, pendingFileIdx: IntArray, expectedBytes: Long,
-        cacheBudget: Long, caBundlePath: String, maxWorkers: Int, processWorkers: Int, listener: StreamListener,
+        cacheBudget: Long, overlapBudget: Long, caBundlePath: String, maxWorkers: Int, processWorkers: Int, listener: StreamListener,
     ): Long
 
     @JvmStatic

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

@@ -266,12 +266,13 @@ Bannerlator's `BlEpicDownload.run(...)` wraps `nativeStart` in a blocking call t
 
 | Method | JNI signature | Symbol |
 |---|---|---|
-| `nativeStream(...)`: Long | `([BLjava/lang/String;Ljava/lang/String;Ljava/lang/String;[Ljava/lang/String;[IJJLjava/lang/String;IILcom/droiddeck/launcher/stores/epic/EpicNative$StreamListener;)J` | `Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeStream` |
+| `nativeStream(...)`: Long | `([BLjava/lang/String;Ljava/lang/String;Ljava/lang/String;[Ljava/lang/String;[IJJJLjava/lang/String;IILcom/droiddeck/launcher/stores/epic/EpicNative$StreamListener;)J` | `Java_com_droiddeck_launcher_stores_epic_EpicNative_nativeStream` |
 
 ```kotlin
 @JvmStatic private external fun nativeStream(
     manifest: ByteArray, installDir: String, chunkCacheDir: String, stagingDir: String,
-    cdnPrefixes: Array<String>, pendingFileIdx: IntArray, expectedBytes: Long, cacheBudget: Long, caBundlePath: String,
+    cdnPrefixes: Array<String>, pendingFileIdx: IntArray, expectedBytes: Long, cacheBudget: Long,
+    overlapBudget: Long, caBundlePath: String,
     maxWorkers: Int, processWorkers: Int, listener: StreamListener,
 ): Long
 ```
@@ -280,7 +281,11 @@ Parameters as `nativeStart`, plus `stagingDir` (the manager's owned
 `<installDir>/.droiddeck-epic-staging`, which must exist; `""` = stage beside each file), and
 except `expectedBytes` (the manager's `Σ fileSize` of the pending
 files, `-1` = no cross-check; no chunk count) and `cacheBudget` (the most bytes of shared chunks
-the cache may hold; the manager sizes it from free space, `0` = never cache, refetch instead).
+the cache may hold; the manager sizes it from free space, `0` = never cache, refetch instead) and
+`overlapBudget` (the most planned bytes of *replaced* files - destination already holds data -
+staged at once, since each sits beside its old copy until its swap; one is always allowed so the
+run progresses; `-1` = no limit). Files are admitted in order and a chunk is dispatched only once
+the file of its first use is admitted; fresh files are never held back.
 `chunkCacheDir` is `""` (`<installDir>/.chunks`) or a scratch directory; the manager falls back to
 `""` for a card install when internal storage cannot hold the budget.
 

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

@@ -9,7 +9,8 @@
 //!             listener: EpicNative.Listener) -> long handle (0 = not started)
 //! nativeStream(manifest: byte[], installDir: String, chunkCacheDir: String, stagingDir: String,
 //!             cdnPrefixes: String[],
-//!             pendingFileIdx: int[], expectedBytes: long, cacheBudget: long, caBundlePath: String,
+//!             pendingFileIdx: int[], expectedBytes: long, cacheBudget: long, overlapBudget: long,
+//!             caBundlePath: String,
 //!             maxWorkers: int, processWorkers: int, listener: EpicNative.StreamListener)
 //!             -> long handle (0 = not started)
 //! nativeCancel(handle)      — flips the run's AtomicBool; fetch, assembly or stream stops promptly
@@ -521,6 +522,7 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
     pending_file_idx: JIntArray,
     expected_bytes: jlong,
     cache_budget: jlong,
+    overlap_budget: jlong,
     ca_bundle_path: JString,
     max_workers: jint,
     process_workers: jint,
@@ -579,6 +581,12 @@ pub extern "system" fn Java_com_droiddeck_launcher_stores_epic_EpicNative_native
             None
         },
         cache_budget: cache_budget.max(0) as u64,
+        // Negative = no limit (nothing the manager knows of is being replaced).
+        replace_overlap_budget: if overlap_budget < 0 {
+            u64::MAX
+        } else {
+            overlap_budget as u64
+        },
         ca_bundle_path,
         max_workers,
         process_workers,

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

@@ -30,6 +30,18 @@
 //! cached chunk is written into every newly reached file at once, so what stays in the cache is
 //! exactly what the simulation kept at that point of the file order.
 //!
+//! **Updates are paced by a replacement-overlap budget.** A staged file that replaces an existing
+//! one sits beside the old copy until its swap, so every replaced file open at once costs its full
+//! new size in extra space (on a fresh install a staged file simply becomes the file). Files are
+//! admitted in order; a replaced file is admitted only while the planned sizes of admitted,
+//! not-yet-swapped replaced files stay within the budget the manager computed from free space,
+//! and one is always admitted when none is open, so the install always progresses. The fetch core
+//! dispatches a chunk only once the file of its first use is admitted ([`run_fetch_gated`]).
+//! Admission never changes the order files are reached in, only when, so the cache simulation
+//! below (which works in file order) stays exact: the cache holds at any moment what the plan holds
+//! at that point of the order. Nothing waits on a gated file: every chunk an admitted file needs is
+//! first used at or before that file, so it was dispatched before the gate closed.
+//!
 //! **A slow disk throttles the download by itself.** Every write happens in the fetch core's
 //! process pool, inside [`FetchSink::process`]; the core frees a body's share of its in-flight
 //! byte budget only when `process` returns, so while the disk is behind, the budget fills and no
@@ -52,13 +64,13 @@ use std::fs::{self, File, OpenOptions};
 use std::io::ErrorKind;
 use std::os::unix::fs::FileExt;
 use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::Mutex;
 use std::time::Instant;
 
 use sha1::{Digest, Sha1};
 
-use crate::fetch_core::{run_fetch, FetchItem, FetchOptions, FetchSink, SinkError};
+use crate::fetch_core::{run_fetch_gated, FetchItem, FetchOptions, FetchSink, SinkError};
 
 use super::assemble::relative_path;
 use super::chunk::{chunk_window_bytes, decode_verified_chunk, part_path};
@@ -222,6 +234,9 @@ pub struct StreamRequest {
     pub expected_bytes: Option<u64>,
     /// Most bytes of shared chunks the cache may hold.
     pub cache_budget: u64,
+    /// Most planned bytes of replaced files (destination already there) staged at once; one is
+    /// always allowed. Fresh files are not limited by it.
+    pub replace_overlap_budget: u64,
     pub ca_bundle_path: String,
     pub max_workers: usize,
     pub process_workers: usize,
@@ -246,6 +261,8 @@ pub struct PlanFile {
     pub parts: Vec<PlanPart>,
     /// Position of the file's first part in the install's part sequence.
     pub first_use: usize,
+    /// The destination exists with data: staging this file costs space until its swap.
+    pub replaces: bool,
 }
 
 #[derive(Debug)]
@@ -272,6 +289,7 @@ pub struct StreamPlan {
     pub staging_dir: Option<PathBuf>,
     pub bytes_total: u64,
     pub budget: u64,
+    pub overlap_budget: u64,
     pub hosts: Vec<String>,
     pub direct_chunks: usize,
     pub shared_chunks: usize,
@@ -349,7 +367,11 @@ pub fn build_stream_plan(req: &StreamRequest) -> Result<StreamPlan, String> {
             chunk_of_use.push(ci);
         }
         bytes_total += file_offset;
+        let replaces = fs::metadata(Path::new(&req.install_dir).join(&rel_path))
+            .map(|m| m.is_file() && m.len() > 0)
+            .unwrap_or(false);
         files.push(PlanFile {
+            replaces,
             rel_path,
             size: file_offset,
             sha1: f.sha1.filter(|h| h.iter().any(|&b| b != 0)),
@@ -458,6 +480,7 @@ pub fn build_stream_plan(req: &StreamRequest) -> Result<StreamPlan, String> {
         staging_dir,
         bytes_total,
         budget: req.cache_budget,
+        overlap_budget: req.replace_overlap_budget,
         hosts,
         direct_chunks,
         shared_chunks,
@@ -535,6 +558,11 @@ struct Shared {
     reached: usize,
     /// `fetch_events[..next_fetch]` have all been downloaded.
     next_fetch: usize,
+    /// Files `[0, admitted)` passed the replacement-overlap budget; `active` never passes it.
+    admitted: usize,
+    /// Planned bytes of admitted replaced files not swapped in yet.
+    overlap_open: u64,
+    peak_overlap: u64,
     cache_bytes: u64,
     peak_cache: u64,
 }
@@ -585,6 +613,8 @@ pub struct Engine<'a> {
     replace_gate: Mutex<Option<Result<(), String>>>,
     stats: Stats,
     started: Instant,
+    /// `Shared::admitted`, readable by the fetch driver without the lock.
+    admitted: AtomicUsize,
 }
 
 fn fatal_write(rel: &str, err: impl std::fmt::Display) -> SinkError {
@@ -629,6 +659,9 @@ impl<'a> Engine<'a> {
                 active: 0,
                 reached: 0,
                 next_fetch: 0,
+                admitted: 0,
+                overlap_open: 0,
+                peak_overlap: 0,
                 cache_bytes,
                 peak_cache: cache_bytes,
             }),
@@ -636,18 +669,45 @@ impl<'a> Engine<'a> {
             replace_gate: Mutex::new(None),
             stats: Stats::default(),
             started: Instant::now(),
+            admitted: AtomicUsize::new(0),
+        }
+    }
+
+    /// The fetch driver's gate: fetch item `id` may go out once its first file is admitted.
+    pub fn may_dispatch(&self, id: usize) -> bool {
+        let p = self.plan;
+        match p.fetch_events.get(id) {
+            Some(&e) => p.event_first_file[e] < self.admitted.load(Ordering::Acquire),
+            None => true,
+        }
+    }
+
+    /// The most planned bytes of replaced files that were staged at once.
+    pub fn peak_replace_overlap(&self) -> u64 {
+        self.shared.lock().map(|st| st.peak_overlap).unwrap_or_default()
+    }
+
+    /// Keep reaching files until nothing new can be written: a file that finishes can admit the
+    /// next replaced file, whose parts may already be in the cache. Iterative, so a long run of
+    /// cache-only files never deepens the stack.
+    fn settle(&self) -> Result<(), SinkError> {
+        loop {
+            let mut work = Work::default();
+            {
+                let mut st = self.shared.lock().expect("stream state poisoned");
+                self.advance(&mut st, &mut work);
+            }
+            if work.from_cache.is_empty() && work.empty_files.is_empty() {
+                return Ok(());
+            }
+            self.run_work(work)?;
         }
     }
 
     /// Before the first fetch: the files before the first chunk to download can be written from
     /// the cache an earlier run left (or are empty).
     pub fn start(&self) -> Result<(), SinkError> {
-        let mut work = Work::default();
-        {
-            let mut st = self.shared.lock().expect("stream state poisoned");
-            self.advance(&mut st, &mut work);
-        }
-        self.run_work(work)
+        self.settle()
     }
 
     /// Reach every file that can be written now: all files before the first chunk still to
@@ -655,6 +715,18 @@ impl<'a> Engine<'a> {
     /// of newly reached files whose chunk is in the cache are claimed for writing.
     fn advance(&self, st: &mut Shared, work: &mut Work) {
         let p = self.plan;
+        while st.admitted < p.files.len() {
+            let file = &p.files[st.admitted];
+            if file.replaces {
+                if st.overlap_open > 0 && st.overlap_open + file.size > p.overlap_budget {
+                    break;
+                }
+                st.overlap_open += file.size;
+                st.peak_overlap = st.peak_overlap.max(st.overlap_open);
+            }
+            st.admitted += 1;
+        }
+        self.admitted.store(st.admitted, Ordering::Release);
         while st.next_fetch < p.fetch_events.len()
             && st.state[p.fetch_events[st.next_fetch]] != EventState::Pending
         {
@@ -664,7 +736,7 @@ impl<'a> Engine<'a> {
             Some(&e) => p.event_first_file[e],
             None => p.files.len(),
         };
-        let target = st.active.max(frontier).max(st.reached).min(p.files.len());
+        let target = st.active.max(frontier).max(st.reached).min(st.admitted);
         for f in st.active..target {
             let file = &p.files[f];
             if file.parts.is_empty() {
@@ -746,6 +818,7 @@ impl<'a> Engine<'a> {
             }
             self.write_from_memory(&data, &more)?;
         }
+        self.settle()?;
         Ok(chunk.credit_bytes())
     }
 
@@ -934,6 +1007,10 @@ impl<'a> Engine<'a> {
             .fetch_sub(staged.min(self.stats.staging_now.load(Ordering::Relaxed)), Ordering::Relaxed);
         if result.is_ok() {
             let size = self.plan.files[f].size;
+            if self.plan.files[f].replaces {
+                let mut st = self.shared.lock().expect("stream state poisoned");
+                st.overlap_open = st.overlap_open.saturating_sub(size);
+            }
             self.stats.files_done.fetch_add(1, Ordering::Relaxed);
             self.stats.files_bytes.fetch_add(size, Ordering::Relaxed);
             self.report(true);
@@ -1036,13 +1113,13 @@ impl<'a> Engine<'a> {
     pub fn summary(&self) -> String {
         let secs = self.started.elapsed().as_secs_f64().max(0.001);
         let written = self.stats.written.load(Ordering::Relaxed);
-        let peak_cache = self
+        let (peak_cache, peak_overlap) = self
             .shared
             .lock()
-            .map(|st| st.peak_cache)
+            .map(|st| (st.peak_cache, st.peak_overlap))
             .unwrap_or_default();
         format!(
-            "stream-summary mode=adaptive files={}/{} bytes_written={written} direct_bytes={} from_cache_bytes={} cache_stored_bytes={} peak_cache_bytes={peak_cache} budget={} plan_peak_cache_bytes={} peak_staging_bytes={} fetched_chunks={} refetches={} elapsed={secs:.1}s avg_MBps={:.2}",
+            "stream-summary mode=adaptive files={}/{} bytes_written={written} direct_bytes={} from_cache_bytes={} cache_stored_bytes={} peak_cache_bytes={peak_cache} budget={} plan_peak_cache_bytes={} peak_staging_bytes={} peak_replace_overlap_bytes={peak_overlap} overlap_budget={} fetched_chunks={} refetches={} elapsed={secs:.1}s avg_MBps={:.2}",
             self.stats.files_done.load(Ordering::Relaxed),
             self.plan.files.len(),
             self.stats.direct.load(Ordering::Relaxed),
@@ -1051,6 +1128,7 @@ impl<'a> Engine<'a> {
             self.plan.budget,
             self.plan.schedule.peak_cache_bytes,
             self.stats.staging_peak.load(Ordering::Relaxed),
+            self.plan.overlap_budget,
             self.stats.fetched.load(Ordering::Relaxed),
             self.plan.refetches,
             written as f64 / secs / (1024.0 * 1024.0),
@@ -1101,7 +1179,7 @@ pub fn run_stream(
     let log = hooks.log;
     let host_cap = per_host_cap(req.max_workers, plan.hosts.len());
     log(&format!(
-        "stream-plan chunk_dir={} cache_dir={} scratch={} {} files={} bytes={} uses={} chunks_direct={} chunks_shared={} fetches={} preloaded={} refetches={} budget={} plan_peak_cache={} hosts={} workers={} per_host_cap={host_cap} process_workers={}",
+        "stream-plan chunk_dir={} cache_dir={} scratch={} {} files={} bytes={} uses={} chunks_direct={} chunks_shared={} fetches={} preloaded={} refetches={} budget={} plan_peak_cache={} replaced_files={} overlap_budget={} hosts={} workers={} per_host_cap={host_cap} process_workers={}",
         plan.manifest.chunk_dir,
         plan.cache_dir.display(),
         plan.scratch_cache,
@@ -1116,6 +1194,8 @@ pub fn run_stream(
         plan.refetches,
         plan.budget,
         plan.schedule.peak_cache_bytes,
+        plan.files.iter().filter(|f| f.replaces).count(),
+        plan.overlap_budget,
         plan.hosts.len(),
         req.max_workers,
         req.process_workers
@@ -1167,7 +1247,17 @@ pub fn run_stream(
         // Progress is the engine's (file bytes written); the core's own credit only feeds its
         // `fetch-end` line.
         let no_progress = |_: u64, _: u64| {};
-        let fetched = run_fetch(items, &plan.hosts, &opts, &engine, cancel, &no_progress, log);
+        let gate = |id: usize| engine.may_dispatch(id);
+        let fetched = run_fetch_gated(
+            items,
+            &plan.hosts,
+            &opts,
+            &engine,
+            cancel,
+            &no_progress,
+            log,
+            &gate,
+        );
         error = fetched.error;
     }
 
@@ -1355,6 +1445,7 @@ mod tests {
                 pending_file_indices: pending,
                 expected_bytes: None,
                 cache_budget: budget,
+                replace_overlap_budget: u64::MAX,
                 ca_bundle_path: String::new(),
                 max_workers: 4,
                 process_workers: 2,
@@ -1664,6 +1755,192 @@ mod tests {
         assert!(plan.reserved.contains(&fx.install.join("a.bin")));
     }
 
+    // ── replacement-overlap gate ──
+
+    /// `n` files of `len` bytes, one chunk each.
+    fn big_fixture(tag: &str, n: usize, len: usize) -> Fixture {
+        let dir = temp_dir(tag);
+        let install = dir.join("game");
+        fs::create_dir_all(&install).unwrap();
+        let guid = |i: usize| [0x100 + i as u32, 0, 0, 0];
+        Fixture {
+            dir,
+            install,
+            chunks: (0..n).map(|i| (guid(i), data(i as u32 + 10, len))).collect(),
+            files: (0..n)
+                .map(|i| (format!("f{i}.bin"), vec![(guid(i), 0, len as i32)]))
+                .collect(),
+        }
+    }
+
+    /// Drive the engine the way the fetch core does: items in order while the gate allows,
+    /// bodies handed back newest first (the most out-of-order the core can be).
+    fn drive_gated(fx: &Fixture, plan: &StreamPlan, engine: &Engine) -> Result<(), SinkError> {
+        engine.start()?;
+        let mut next = 0;
+        let mut in_flight: Vec<usize> = Vec::new();
+        while next < plan.fetch_events.len() || !in_flight.is_empty() {
+            while next < plan.fetch_events.len() && engine.may_dispatch(next) {
+                in_flight.push(next);
+                next += 1;
+            }
+            let id = in_flight.pop().expect("gate closed with nothing in flight: deadlock");
+            engine.process(&item(id), fx.body_for(plan, id))?;
+        }
+        Ok(())
+    }
+
+    fn write_old(fx: &Fixture, files: &[usize]) {
+        for &i in files {
+            fs::write(fx.install.join(&fx.files[i].0), b"old").unwrap();
+        }
+    }
+
+    #[test]
+    fn replaced_files_stay_within_the_overlap_budget_and_complete() {
+        let fx = big_fixture("stream-overlap", 6, 1000);
+        write_old(&fx, &[0, 1, 2, 3, 4, 5]);
+        let mut req = fx.request((0..6).collect(), 1 << 20);
+        req.replace_overlap_budget = 2500;
+        let plan = build_stream_plan(&req).unwrap();
+        assert!(plan.files.iter().all(|f| f.replaces));
+        let cancel = AtomicBool::new(false);
+        let rec = Recorder::new();
+        with_engine(&plan, &cancel, &rec, |engine| {
+            engine.start().unwrap();
+            assert!(engine.may_dispatch(1), "two 1000-byte files fit 2500");
+            assert!(!engine.may_dispatch(2), "a third does not");
+            drive_gated(&fx, &plan, engine).unwrap();
+            assert_eq!(engine.peak_replace_overlap(), 2000);
+            assert_eq!(engine.unfinished_files(), 0);
+        });
+        for i in 0..6 {
+            assert_eq!(fs::read(fx.install.join(&fx.files[i].0)).unwrap(), fx.expected(i));
+        }
+        assert!(fx.stages().is_empty());
+    }
+
+    #[test]
+    fn a_budget_below_the_largest_file_still_progresses_one_at_a_time() {
+        let fx = big_fixture("stream-overlap-one", 4, 800);
+        write_old(&fx, &[0, 1, 2, 3]);
+        let mut req = fx.request((0..4).collect(), 1 << 20);
+        req.replace_overlap_budget = 1;
+        let plan = build_stream_plan(&req).unwrap();
+        let cancel = AtomicBool::new(false);
+        let rec = Recorder::new();
+        with_engine(&plan, &cancel, &rec, |engine| {
+            engine.start().unwrap();
+            assert!(engine.may_dispatch(0));
+            assert!(!engine.may_dispatch(1));
+            drive_gated(&fx, &plan, engine).unwrap();
+            assert_eq!(engine.peak_replace_overlap(), 800);
+            assert_eq!(engine.unfinished_files(), 0);
+        });
+        for i in 0..4 {
+            assert_eq!(fs::read(fx.install.join(&fx.files[i].0)).unwrap(), fx.expected(i));
+        }
+    }
+
+    #[test]
+    fn a_fresh_install_is_never_gated() {
+        let fx = big_fixture("stream-overlap-fresh", 5, 600);
+        let mut req = fx.request((0..5).collect(), 1 << 20);
+        req.replace_overlap_budget = 1;
+        let plan = build_stream_plan(&req).unwrap();
+        assert!(plan.files.iter().all(|f| !f.replaces));
+        let cancel = AtomicBool::new(false);
+        let rec = Recorder::new();
+        with_engine(&plan, &cancel, &rec, |engine| {
+            engine.start().unwrap();
+            assert!((0..5).all(|id| engine.may_dispatch(id)));
+            drive_gated(&fx, &plan, engine).unwrap();
+            assert_eq!(engine.peak_replace_overlap(), 0);
+        });
+        assert_eq!(fs::read(fx.install.join("f4.bin")).unwrap(), fx.expected(4));
+    }
+
+    #[test]
+    fn mixed_fresh_and_replaced_files_gate_only_the_replacements() {
+        let fx = big_fixture("stream-overlap-mixed", 6, 700);
+        write_old(&fx, &[1, 3, 5]);
+        let mut req = fx.request((0..6).collect(), 1 << 20);
+        req.replace_overlap_budget = 700;
+        let plan = build_stream_plan(&req).unwrap();
+        let cancel = AtomicBool::new(false);
+        let rec = Recorder::new();
+        with_engine(&plan, &cancel, &rec, |engine| {
+            engine.start().unwrap();
+            // f0 fresh, f1 replaced (admitted), f2 fresh, f3 replaced (waits for f1's swap).
+            assert!(engine.may_dispatch(2));
+            assert!(!engine.may_dispatch(3));
+            drive_gated(&fx, &plan, engine).unwrap();
+            assert_eq!(engine.peak_replace_overlap(), 700);
+            assert_eq!(engine.unfinished_files(), 0);
+        });
+        for i in 0..6 {
+            assert_eq!(fs::read(fx.install.join(&fx.files[i].0)).unwrap(), fx.expected(i));
+        }
+    }
+
+    #[test]
+    fn a_shared_chunk_for_a_gated_file_waits_in_the_cache() {
+        // f0 and f2 are replaced and share chunk A; f1 is fresh. With room for one replaced file,
+        // f2 is admitted only after f0's swap, and A is cached until then.
+        let dir = temp_dir("stream-overlap-shared");
+        let install = dir.join("game");
+        fs::create_dir_all(&install).unwrap();
+        let fx = Fixture {
+            dir,
+            install,
+            chunks: vec![(A, data(1, 300)), (B, data(2, 50))],
+            files: vec![
+                ("f0.bin".to_string(), vec![(A, 0, 100)]),
+                ("f1.bin".to_string(), vec![(B, 0, 50)]),
+                ("f2.bin".to_string(), vec![(A, 100, 200)]),
+            ],
+        };
+        write_old(&fx, &[0, 2]);
+        let mut req = fx.request(vec![0, 1, 2], 1 << 20);
+        req.replace_overlap_budget = 150;
+        let plan = build_stream_plan(&req).unwrap();
+        let cancel = AtomicBool::new(false);
+        let rec = Recorder::new();
+        with_engine(&plan, &cancel, &rec, |engine| {
+            drive_gated(&fx, &plan, engine).unwrap();
+            assert!(engine.peak_replace_overlap() <= 200);
+            assert_eq!(engine.unfinished_files(), 0);
+        });
+        for i in 0..3 {
+            assert_eq!(fs::read(fx.install.join(&fx.files[i].0)).unwrap(), fx.expected(i));
+        }
+        assert_eq!(fs::read_dir(&plan.cache_dir).unwrap().count(), 0);
+    }
+
+    #[test]
+    fn cancelling_a_gated_update_keeps_every_old_file() {
+        let fx = big_fixture("stream-overlap-cancel", 4, 900);
+        write_old(&fx, &[0, 1, 2, 3]);
+        let mut req = fx.request((0..4).collect(), 1 << 20);
+        req.replace_overlap_budget = 1000;
+        let plan = build_stream_plan(&req).unwrap();
+        let cancel = AtomicBool::new(false);
+        let rec = Recorder::new();
+        with_engine(&plan, &cancel, &rec, |engine| {
+            engine.start().unwrap();
+            engine.process(&item(0), fx.body_for(&plan, 0)).unwrap(); // f0 swapped in
+            assert!(engine.may_dispatch(1));
+            cancel.store(true, Ordering::Relaxed);
+            assert!(engine.process(&item(1), fx.body_for(&plan, 1)).is_err());
+            engine.discard_staging();
+        });
+        assert_eq!(fs::read(fx.install.join("f0.bin")).unwrap(), fx.expected(0));
+        for i in 1..4 {
+            assert_eq!(fs::read(fx.install.join(&fx.files[i].0)).unwrap(), b"old");
+        }
+        assert!(fx.stages().is_empty());
+    }
+
     fn item(id: usize) -> FetchItem {
         FetchItem {
             id: id as u64,

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

@@ -221,6 +221,25 @@ pub fn run_fetch(
     cancel: &AtomicBool,
     progress: &(dyn Fn(u64, u64) + Sync),
     log: &(dyn Fn(&str) + Sync),
+) -> FetchOutcome {
+    run_fetch_gated(items, hosts, opts, sink, cancel, progress, log, &|_: usize| true)
+}
+
+/// [`run_fetch`] with an admission gate: the next fresh item (by index) is only dispatched once
+/// `gate(index)` says so. Items are still taken strictly in order, and retries are never held
+/// back. While the gate is closed the driver idles as it does on a full window, so a gate that
+/// opens as the sink makes progress (e.g. as files finish) paces the run without stalling the
+/// requests already in flight. The adaptivity is unchanged.
+#[allow(clippy::too_many_arguments)]
+pub fn run_fetch_gated(
+    items: Vec<FetchItem>,
+    hosts: &[String],
+    opts: &FetchOptions,
+    sink: &dyn FetchSink,
+    cancel: &AtomicBool,
+    progress: &(dyn Fn(u64, u64) + Sync),
+    log: &(dyn Fn(&str) + Sync),
+    gate: &(dyn Fn(usize) -> bool + Sync),
 ) -> FetchOutcome {
     let started = Instant::now();
     let label = opts.label.as_str();
@@ -484,6 +503,7 @@ mode={}",
             Ok(rt) => {
                 rt.block_on(run_driver(DriverCtx {
                     items: items.as_slice(),
+                    gate,
                     hosts: hosts.as_slice(),
                     opts,
                     label,
@@ -1006,6 +1026,7 @@ struct FetchDone {
 
 struct DriverCtx<'a> {
     items: &'a [FetchItem],
+    gate: &'a (dyn Fn(usize) -> bool + Sync),
     hosts: &'a [String],
     opts: &'a FetchOptions,
     label: &'a str,
@@ -1027,6 +1048,7 @@ struct DriverCtx<'a> {
 async fn run_driver(ctx: DriverCtx<'_>) {
     let DriverCtx {
         items,
+        gate,
         hosts,
         opts,
         label,
@@ -1182,10 +1204,10 @@ async fn run_driver(ctx: DriverCtx<'_>) {
             let (idx, attempts) = if use_retry {
                 let p = *retry.front().expect("retry nonempty");
                 (p.idx, p.attempts)
-            } else if next_item < items.len() {
+            } else if next_item < items.len() && gate(next_item) {
                 (next_item, 0u32)
             } else {
-                break; // nothing dispatchable now (retry not due, or all dispatched)
+                break; // nothing dispatchable now (retry not due, gate closed, or all dispatched)
             };
             let item = &items[idx];
 

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

@@ -1,7 +1,10 @@
 package com.droiddeck.launcher.stores.epic
 
+import java.io.File
 import org.junit.Assert.*
+import org.junit.Rule
 import org.junit.Test
+import org.junit.rules.TemporaryFolder
 import org.junit.runner.RunWith
 import org.robolectric.RobolectricTestRunner
 import org.robolectric.annotation.Config
@@ -9,14 +12,29 @@ import org.robolectric.annotation.Config
 @RunWith(RobolectricTestRunner::class)
 @Config(sdk = [33])
 class EpicAdaptivePlanTest {
+    @get:Rule val temp = TemporaryFolder()
     private val mib = 1024L * 1024
     private val gib = 1024 * mib
 
-    @Test fun freshInstallsNeedNoOpenStagingAllowance() {
-        assertEquals(0L, EpicDownloadManager.adaptiveOpenStagingAllowance(emptyMap()))
-        val old = mapOf("big.pak" to 10 * gib, "small.pak" to gib)
-        assertEquals(10 * gib + EpicDownloadManager.ADAPTIVE_OPEN_WINDOW_BYTES, EpicDownloadManager.adaptiveOpenStagingAllowance(old))
-        assertEquals(100 * mib, EpicDownloadManager.adaptiveOpenStagingAllowance(mapOf("a" to 60 * mib, "b" to 40 * mib)))
+    private fun file(name: String, size: Long) = EpicDownloadManager.FileInfo().apply {
+        filename = name
+        var left = size
+        while (left > 0) {
+            val part = minOf(left, Int.MAX_VALUE.toLong())
+            parts.add(EpicDownloadManager.ChunkPart().apply { this.size = part.toInt() })
+            left -= part
+        }
+    }
+
+    @Test fun netPeakCreditsEachOldCopyAtItsSwapAndLargestReplacedIgnoresFreshFiles() {
+        val files = listOf(file("a", 10 * gib), file("b", 2 * gib), file("c", 4 * gib))
+        val old = mapOf("a" to 9 * gib, "c" to 5 * gib)
+        // Running Σ(new − old): +1, +3, +2 GiB.
+        assertEquals(3 * gib, EpicDownloadManager.settledNetPeak(files, old))
+        assertEquals(10 * gib, EpicDownloadManager.largestReplacedFile(files, old))
+        assertEquals(16 * gib, EpicDownloadManager.settledNetPeak(files, emptyMap()))
+        assertEquals(0L, EpicDownloadManager.largestReplacedFile(files, emptyMap()))
+        assertEquals(0L, EpicDownloadManager.largestReplacedFile(files, mapOf("b" to 0L)))
     }
 
     @Test fun cacheBudgetIsHalfTheHeadroomCappedAndDefaultsWhenSpaceIsUnknown() {
@@ -27,13 +45,39 @@ class EpicAdaptivePlanTest {
         assertEquals((free - 400 * mib - EpicDownloadManager.ADAPTIVE_TRANSIENT_BYTES) / 2, EpicDownloadManager.adaptiveCacheBudget(free, 400 * mib))
     }
 
-    @Test fun cardInstallKeepsTheCacheOnInternalScratchWhenItFits() {
+    @Test fun freshInstallsHaveNoOverlapLimit() {
         val plan = EpicDownloadManager.planAdaptiveSpace(20 * gib, 0, 30 * gib, true, 10 * gib)
         assertNull(plan.error)
-        assertFalse(plan.cacheBesideGame)
-        assertEquals(EpicDownloadManager.ADAPTIVE_CACHE_MAX_BYTES, plan.cacheBudget)
+        assertEquals(-1L, plan.overlapBudget)
         assertEquals(20 * gib, plan.gameNeed)
-        assertEquals(plan.cacheBudget + EpicDownloadManager.ADAPTIVE_TRANSIENT_BYTES, plan.scratchNeed)
+    }
+
+    @Test fun updateOverlapBudgetIsTheLeftoverAfterNetGrowthAndCache() {
+        // Card install, cache on internal: the card's leftover after the net growth is the budget.
+        val roomy = EpicDownloadManager.planAdaptiveSpace(gib, 4 * gib, 20 * gib, true, 10 * gib)
+        assertNull(roomy.error)
+        assertFalse(roomy.cacheBesideGame)
+        assertEquals(19 * gib, roomy.overlapBudget)
+        assertEquals(20 * gib, roomy.gameNeed)
+        // Tight: floored at the largest replaced file, one at a time.
+        val tight = EpicDownloadManager.planAdaptiveSpace(gib, 4 * gib, 5 * gib, true, 10 * gib)
+        assertNull(tight.error)
+        assertEquals(4 * gib, tight.overlapBudget)
+        // Not even one replaced file fits: the shortfall is named.
+        val short = EpicDownloadManager.planAdaptiveSpace(gib, 4 * gib, 4 * gib, true, 10 * gib)
+        assertTrue(short.error!!.contains("game's storage"))
+        assertTrue(short.error!!.contains("short by 1.0 GB"))
+        // Unknown free space: one replaced file at a time.
+        assertEquals(4 * gib, EpicDownloadManager.planAdaptiveSpace(gib, 4 * gib, -1, true, -1).overlapBudget)
+    }
+
+    @Test fun cacheBesideTheGameComesOutOfTheOverlapBudget() {
+        val plan = EpicDownloadManager.planAdaptiveSpace(gib, 2 * gib, 10 * gib, false, -1)
+        assertNull(plan.error)
+        assertTrue(plan.cacheBesideGame)
+        assertTrue(plan.cacheBudget > 0)
+        assertEquals(10 * gib - gib - plan.cacheBudget - EpicDownloadManager.ADAPTIVE_TRANSIENT_BYTES, plan.overlapBudget)
+        assertEquals(10 * gib, plan.gameNeed)
     }
 
     @Test fun fullInternalStorageMovesTheCacheBesideTheGameInsteadOfFailing() {
@@ -43,15 +87,21 @@ class EpicAdaptivePlanTest {
         assertEquals(0L, plan.scratchNeed)
         assertTrue(plan.cacheBudget > 0)
         assertEquals(20 * gib + plan.cacheBudget + EpicDownloadManager.ADAPTIVE_TRANSIENT_BYTES, plan.gameNeed)
-        assertTrue(plan.gameNeed <= 30 * gib)
     }
 
-    @Test fun shortStorageIsNamedWithTheShortfall() {
-        val card = EpicDownloadManager.planAdaptiveSpace(20 * gib, 0, 10 * gib, true, 10 * gib)
-        assertTrue(card.error!!.contains("game's storage"))
-        assertTrue(card.error!!.contains("short by 10.0 GB"))
-        val internal = EpicDownloadManager.planAdaptiveSpace(20 * gib, gib, 20 * gib, false, -1)
-        assertTrue(internal.error!!.contains("game's storage"))
-        assertNull(EpicDownloadManager.planAdaptiveSpace(20 * gib, 0, -1, true, -1).error)
+    @Test fun ownedStagingIsRemovedOnlyWhenItHoldsJustItsMarker() {
+        val root = temp.newFolder("game")
+        val staging = EpicDownloadManager.recoverOwnedStaging(root, emptyList())
+        File(staging, "keep.stage").writeText("x")
+        EpicDownloadManager.releaseOwnedStaging(root)
+        assertTrue(File(staging, "owner").exists())
+        File(staging, "keep.stage").delete()
+        EpicDownloadManager.releaseOwnedStaging(root)
+        assertFalse(staging.exists())
+        // A directory without the marker is not ours.
+        staging.mkdir()
+        File(staging, "owner").writeText("someone else")
+        EpicDownloadManager.releaseOwnedStaging(root)
+        assertTrue(staging.exists())
     }
 }

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

@@ -578,7 +578,8 @@ class EpicBoundedInstallTest {
         ShadowStatFs.registerStats(root.path, 1024, peak / 4096, peak / 4096)
         assertNotNull(install())
         assertEquals("keep", legitimate.readText())
-        assertEquals(listOf("owner"), staging.listFiles()!!.map { it.name })
+        // The orphan was recovered, and a finished install removes the then-empty owned namespace.
+        assertFalse(staging.exists())
     }
 
     @Test fun stagingRecoveryRefusesUnownedAndManifestOwnedNamespaces() {

docs/development/stores.md

@@ -71,12 +71,25 @@ The run logs `stream-plan` (direct vs shared chunks, planned refetches, budget)
 `stream-summary mode=adaptive` (peak cache, peak staging, refetches, bytes written from the
 download vs from the cache).
 
+On an update, a staged file that replaces an existing one sits beside its old copy until its swap,
+so the engine paces replacements with an overlap budget: files are admitted in order, a replaced
+file only while the planned sizes of admitted, not-yet-swapped replaced files fit the budget (one
+is always admitted, so the run progresses), and a chunk is fetched only once its file is admitted.
+Fresh files are never held back, and admission only delays files, never reorders them, so the
+cache plan stays exact. The run logs `overlap_budget=` in `stream-plan` and
+`peak_replace_overlap_bytes=` in `stream-summary`. With room to spare the budget is large and
+updates run as wide as fresh installs; on a nearly full volume, replaced files are staged one
+after another and the fetch pauses at each replaced-file boundary until the previous one is
+swapped in.
+
 Its space plan (`EpicDownloadManager.planAdaptiveSpace`, unit-tested in `EpicAdaptivePlanTest`)
-checks each volume for what lands on it. The game volume: the sequential staged-file peak
-(remaining file bytes, each replaced file's old copy credited only after its swap), plus, for an
-update, the largest old file plus 256 MiB for files finishing out of order within the fetch
-window, plus the cache when it is beside the game. The cache volume: the budget plus 64 MiB of
-chunks in transit. The budget is half the cache volume's free space after those needs, at most
+checks each volume for what lands on it, with the lookahead planner's rule (the need must not
+exceed the free reading). The game volume: the net growth at its worst point (`settledNetPeak`,
+the running Σ(new − old) with each old copy released at its swap), plus the overlap budget, plus
+the cache when it is beside the game. The overlap budget is whatever the game volume has left
+after the net growth and the cache, never less than the largest replaced file; when even that
+does not fit, the plan hands over to the lookahead engine. The cache volume: the budget plus 64 MiB of
+chunks in transit. The cache budget is half the cache volume's free space after those needs, at most
 1 GiB (256 MiB when free space is unknown; 0 is allowed and means every reuse is a refetch). On a
 card install the cache stays on internal scratch unless internal storage cannot hold at least
 64 MiB of it; then the cache goes beside the game (still released chunk by chunk) instead of
@@ -158,7 +171,8 @@ are checked separately. File SHA-1 hashes are checked where supplied by the mani
 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.
 All three staged writers (adaptive, lookahead, Java) use `<installDir>/.droiddeck-epic-staging/`, whose `owner` marker identifies this
-namespace. Before the delta pass and free-space probes, the manager removes orphan `.stage`
+namespace. A successful install removes it when it holds nothing but the marker (same no-follow
+and canonical-path checks); the next run recreates it. 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.