Merge pull request 'agent: stop downloader pool stampeding a slow curator (congestion collapse)' (#177) from dev into main

agent/fc_agent/app.py

@@ -17,7 +17,7 @@ from .worker import Worker
 # Bump on every agent change. The page embeds this and /status reports it; the UI
 # warns to reload when they differ — so a stale browser-cached page can't be
 # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.)
-VERSION = "2026-07-01.3 · video frame dedup"
+VERSION = "2026-07-01.4 · gentler downloads, failure-aware scaling"
 
 logbuf.install()
 cfg = Config.from_env()

agent/fc_agent/client.py

@@ -113,7 +113,11 @@ class FcClient:
 
     def fetch_image(self, image_url: str) -> bytes:
         # image_url is a server-relative path ("/images/...").
-        r = self.s.get(f"{self.base}{image_url}", timeout=180)
+        # timeout=(connect, read): the read timeout is BETWEEN-BYTES, not total,
+        # so a large-but-flowing download still completes — but a stuck/dead
+        # connection (curator overloaded) fails in 60s instead of hanging a
+        # downloader for 180s and piling up concurrent stuck requests on curator.
+        r = self.s.get(f"{self.base}{image_url}", timeout=(10, 60))
         r.raise_for_status()
         return r.content
 

agent/fc_agent/worker.py

@@ -61,10 +61,14 @@ def _is_transient(exc: requests.RequestException) -> bool:
     return resp.status_code >= 500 or resp.status_code in (401, 403, 408, 409, 429)
 
 
-# Pipeline sizing. Downloaders are I/O-bound so the ceiling is generous; consumers
-# are GPU-bound so a couple saturate the card. The buffer is small on purpose —
-# each slot can hold many decoded video frames, so it bounds RAM, not just depth.
-DL_MAX = 24              # max downloader threads
+# Pipeline sizing. Downloaders are I/O-bound, but every download streams a full
+# original (large videos included) THROUGH curator's single Python file-serving
+# path — so the ceiling is deliberately modest: too many concurrent large-file
+# GETs saturate curator's web workers + NFS and slow everything (including the
+# browser). 8 keeps a fast GPU fed without stampeding curator. Consumers are
+# GPU-bound so a couple saturate the card. The buffer is small on purpose — each
+# slot can hold many decoded video frames, so it bounds RAM, not just depth.
+DL_MAX = 8               # max downloader threads
 CONSUMER_MAX = 2         # max GPU consumer threads
 BUFFER_MAX = 12          # bounded decoded-frame buffer (backpressure + RAM cap)
 
@@ -712,13 +716,22 @@ class Worker:
         (the GPU is the bottleneck → add a 2nd consumer if it has headroom and the
         add lifts throughput, else trim a downloader). Occupancy, util and
         throughput are EWMA-smoothed and decisions spaced so moves ride averaged
-        signals, not tick-to-tick noise. VRAM pressure sheds a consumer at once."""
+        signals, not tick-to-tick noise. VRAM pressure sheds a consumer at once.
+
+        Failure guard (critical): an empty buffer can mean the GPU is starving OR
+        that downloads are FAILING (curator slow/unreachable). In the latter case
+        adding downloaders piles more concurrent large-file requests onto a
+        struggling curator — a congestion collapse that slows curator (and its
+        browser) further and never recovers. So if transient download failures
+        rose since the last decision, SHRINK toward the floor instead of growing,
+        and let the pool ramp back up only once downloads succeed again."""
         from . import gpu as gpumod
 
         occ_ewma: float | None = None
         util_ewma: float | None = None
         tput_ewma: float | None = None
         prev_p, prev_t = self.processed, time.monotonic()
+        prev_fail = self.transient
         tick = 0
         con_grew = False         # did the previous decision add a consumer?
         tput_before = 0.0        # smoothed jobs/s before that consumer add
@@ -726,6 +739,7 @@ class Worker:
             if not (self._running and self._auto):
                 occ_ewma = util_ewma = tput_ewma = None
                 prev_p, prev_t = self.processed, time.monotonic()
+                prev_fail = self.transient
                 tick = 0
                 con_grew = False
                 self._util_smooth = None
@@ -764,9 +778,17 @@ class Worker:
             tput_ewma = inst if tput_ewma is None else (
                 TPUT_ALPHA * inst + (1 - TPUT_ALPHA) * tput_ewma
             )
+            fail_delta = self.transient - prev_fail
+            prev_fail = self.transient
 
             d0, c0 = self._dl_target, self._consumer_target
-            if occ_ewma < OCC_LOW:
+            if fail_delta > 0:
+                # Downloads are FAILING (curator slow/unreachable), so the empty
+                # buffer is NOT the GPU starving — growing would stampede a
+                # struggling curator. Back off toward the floor and let it recover.
+                self._apply_downloaders(-1)
+                con_grew = False
+            elif occ_ewma < OCC_LOW:
                 # Buffer starving → GPU idle waiting on downloads → add a feeder.
                 self._apply_downloaders(+1)
                 con_grew = False