From 459ed1a38ed8ebd2ec705a95567a93811899aa02 Mon Sep 17 00:00:00 2001
From: imSp4rky <george+slight@slight.me>
Date: Fri, 10 Jul 2026 13:16:30 -0600
Subject: [PATCH] fix(downloader): release hedge loser file handles before
 merge sweep

superseded racers now exit on the next network arrival and the pool
waits for them on clean completion, so Windows can delete stray .!dev
files at merge time (WinError 32)
---
 unshackle/core/downloaders/requests.py | 23 +++++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/unshackle/core/downloaders/requests.py b/unshackle/core/downloaders/requests.py
index 56b51335..cc2c84aa 100644
--- a/unshackle/core/downloaders/requests.py
+++ b/unshackle/core/downloaders/requests.py
@@ -29,6 +29,9 @@ READ_TIMEOUT = 30
 HEDGE_FACTOR = 3
 HEDGE_MIN_WAIT = 5.0
 
+# racers read per network arrival (read1) so superseded hedge losers exit promptly
+RACER_READ1 = True
+
 # Adaptive chunk sizing — benchmarked optimal range
 MIN_CHUNK = 524_288  # 512KB
 MAX_CHUNK = 4_194_304  # 4MB
@@ -326,7 +329,11 @@ def download(
                 if use_rnet:
                     chunks = stream.stream()
                 elif use_raw:
-                    chunks = iter(lambda: stream.raw.read(chunk_size), b"")
+                    _read1 = getattr(stream.raw, "read1", None) if claimed is not None and RACER_READ1 else None
+                    if _read1 is not None:
+                        chunks = iter(lambda: _read1(chunk_size), b"")
+                    else:
+                        chunks = iter(lambda: stream.raw.read(chunk_size), b"")
                 else:
                     chunks = stream.iter_content(chunk_size=chunk_size)
 
@@ -336,6 +343,13 @@ def download(
                 for chunk in chunks:
                     if DOWNLOAD_CANCELLED.is_set():
                         break
+                    if claimed is not None and claimed():
+                        # close the handle or Windows can't delete the stray .!dev at merge (WinError 32)
+                        try:
+                            stream.close()
+                        except Exception:
+                            pass
+                        return
                     _write(chunk)
                     download_size = len(chunk)
                     written += download_size
@@ -387,6 +401,9 @@ def download(
                 stream.close()
             except Exception:
                 pass
+            if claimed is not None and claimed():
+                # a superseded loser's error must not retry or kill the batch
+                return
             if DOWNLOAD_CANCELLED.is_set() or attempts == MAX_ATTEMPTS:
                 if part_mode and not DOWNLOAD_CANCELLED.is_set():
                     raise
@@ -702,7 +719,9 @@ def requests(
             yield dict(downloaded="[yellow]CANCELLED")
             raise
         finally:
-            pool.shutdown(wait=False, cancel_futures=True)
+            # losers must close their handles before merge sweeps *.!dev (WinError 32);
+            # no wait on cancel/fail: merge never runs and workers may be blocked in reads
+            pool.shutdown(wait=not DOWNLOAD_CANCELLED.is_set(), cancel_futures=True)
 
         # Drain remaining events
         while True: