fix(ffmpeg): give the CF-bypass stream the same retry/resume as aiohttp (#169)

VERSION

@@ -1 +1 @@
-1.8.23
+1.8.24

src/torchlight/FFmpegAudioPlayer.py

@@ -1,4 +1,5 @@
 import asyncio
+import concurrent.futures
 import datetime
 import logging
 import os
@@ -159,38 +160,102 @@ async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str], needs
 
         if needs_cf_bypass:
             queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=10)
+            max_network_retries = 5
+
+            def enqueue_chunk(chunk: bytes) -> bool:
+                # Hand a chunk to the event-loop queue from this worker thread, waking every
+                # 0.5s so a Stop() (self.playing -> False) can't wedge us on a full queue.
+                while self.playing:
+                    future = asyncio.run_coroutine_threadsafe(queue.put(chunk), self.torchlight.loop)
+                    try:
+                        future.result(timeout=0.5)
+                        return True
+                    except concurrent.futures.TimeoutError:
+                        future.cancel()
+                    except Exception as err:
+                        self.logger.debug("curl_cffi enqueue failed: %s", err)
+                        return False
+                return False
 
             def sync_fetch_curl_cffi() -> None:
                 proxies: cffi_requests.ProxySpec | None = {"http": proxy_url, "https": proxy_url} if proxy_url else None
+                bytes_downloaded = 0
                 try:
-                    resp = cffi_requests.get(
-                        uri,
-                        headers=headers,
-                        cookies=cookies,
-                        proxies=proxies,
-                        impersonate="chrome120",
-                        stream=True,
-                        timeout=15,
-                    )
-                    if resp.status_code not in (200, 206):
-                        self.logger.error("HTTP stream failed via curl_cffi with status %d", resp.status_code)
-                        asyncio.run_coroutine_threadsafe(queue.put(None), self.torchlight.loop)
-                        return
+                    # Retry parity with the aiohttp path: resume from bytes_downloaded with a
+                    # Range header instead of restarting the clip from zero on a transient drop.
+                    for attempt in range(1, max_network_retries + 1):
+                        if not self.playing:
+                            return
+
+                        req_headers = dict(headers)
+                        if bytes_downloaded > 0:
+                            req_headers["Range"] = f"bytes={bytes_downloaded}-"
+
+                        try:
+                            resp = cffi_requests.get(
+                                uri,
+                                headers=req_headers,
+                                cookies=cookies,
+                                proxies=proxies,
+                                impersonate="chrome120",
+                                stream=True,
+                                timeout=15,
+                            )
+                        except Exception as err:
+                            self.logger.warning(
+                                "curl_cffi connection failed (%s). Retrying (%d/%d)...",
+                                err,
+                                attempt,
+                                max_network_retries,
+                            )
+                            time.sleep(0.5)
+                            continue
+
+                        try:
+                            if resp.status_code not in (200, 206):
+                                self.logger.error("HTTP stream failed via curl_cffi with status %d", resp.status_code)
+                                return
+
+                            # A plain 200 answering a Range request means the server ignored the
+                            # header and restarted from byte 0; drop what ffmpeg already has.
+                            skip = bytes_downloaded if (bytes_downloaded and resp.status_code == 200) else 0
+                            dropped = False
 
-                    for chunk in resp.iter_content(chunk_size=32 * 1024):
-                        if not chunk or not self.playing:
-                            break
-
-                        while self.playing:
                             try:
-                                future = asyncio.run_coroutine_threadsafe(queue.put(chunk), self.torchlight.loop)
-                                future.result(timeout=0.5)
-                                break
-                            except Exception:
-                                if not self.playing:
-                                    break
-                except Exception as err:
-                    self.logger.error("curl_cffi fetch exception: %s", err)
+                                for chunk in resp.iter_content(chunk_size=32 * 1024):
+                                    if not self.playing:
+                                        return
+                                    if not chunk:
+                                        continue
+
+                                    if skip:
+                                        if len(chunk) <= skip:
+                                            skip -= len(chunk)
+                                            continue
+                                        chunk = chunk[skip:]
+                                        skip = 0
+
+                                    bytes_downloaded += len(chunk)
+                                    if not enqueue_chunk(chunk):
+                                        return
+                            except Exception as err:
+                                dropped = True
+                                self.logger.warning(
+                                    "curl_cffi stream dropped after %d bytes (%s). Retrying (%d/%d)...",
+                                    bytes_downloaded,
+                                    err,
+                                    attempt,
+                                    max_network_retries,
+                                )
+                        finally:
+                            resp.close()
+
+                        if not dropped:
+                            return
+
+                        time.sleep(0.5)
+
+                    self.logger.error("curl_cffi stream gave up after %d attempts", max_network_retries)
                 finally:
                     asyncio.run_coroutine_threadsafe(queue.put(None), self.torchlight.loop)