fix: steady 24fps MJPEG streaming with no frame drops

Restructure stream consumer to drain queue first before blocking on
event, preventing stalls when producer batches frames faster than
consumer reads. Make all queues/deques unbounded so frames are never
silently dropped.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app.py

@@ -551,8 +551,12 @@ async def stream_video(job_id: str):
     import queue as queue_mod
 
     async def stream_generator():
+        STREAM_FPS = 24
+        FRAME_INTERVAL = 1.0 / STREAM_FPS  # ~41.7ms
+
         loop = asyncio.get_running_loop()
         buffered = False
+        last_yield_time = 0.0
 
         # TTFS instrumentation
         _first_yielded = False
@@ -566,48 +570,67 @@ async def stream_video(job_id: str):
         # Get or create the asyncio.Event for this stream (must be in async context)
         event = get_stream_event(job_id)
 
-        while True:
-            q = get_stream(job_id)
-            if not q:
-                break
+        # Hold a local ref to the queue so we can drain it even after remove_stream()
+        q = get_stream(job_id)
+        if not q:
+            return
+
+        stream_removed = False
 
+        while True:
             try:
                 # Initial Buffer: Wait until we have enough frames or job is done
                 if not buffered:
                     if not _buffer_wait_logged and _stream_t0:
                         logging.info("[TTFS:%s] +%.1fs stream_buffer_wait (qsize=%d)", job_id, time.perf_counter() - _stream_t0, q.qsize())
                         _buffer_wait_logged = True
-                    if q.qsize() < 5:
+                    if q.qsize() < 5 and not stream_removed:
                         await asyncio.sleep(0.1)
+                        stream_removed = get_stream(job_id) is None
                         continue
                     buffered = True
                     if _stream_t0:
                         logging.info("[TTFS:%s] +%.1fs stream_buffer_ready", job_id, time.perf_counter() - _stream_t0)
 
-                # Event-driven wait — replaces busy-wait polling
-                if event is not None:
-                    try:
-                        await asyncio.wait_for(event.wait(), timeout=1.0)
-                        event.clear()
-                    except asyncio.TimeoutError:
-                        if not get_stream(job_id):
-                            return
-                        continue
-                else:
-                    # Fallback if no event (shouldn't happen)
-                    await asyncio.sleep(0.033)
-
-                # Drain available frame (already pre-resized by publish_frame)
+                # Try to get a frame from the queue first (non-blocking)
+                frame = None
                 try:
                     frame = q.get_nowait()
                 except queue_mod.Empty:
-                    continue
+                    pass
+
+                # Only block on event when queue is actually empty
+                if frame is None:
+                    if stream_removed:
+                        break  # stream ended and queue fully drained
+                    if event is not None:
+                        try:
+                            await asyncio.wait_for(event.wait(), timeout=1.0)
+                            event.clear()
+                        except asyncio.TimeoutError:
+                            stream_removed = get_stream(job_id) is None
+                            continue
+                    else:
+                        await asyncio.sleep(FRAME_INTERVAL)
+                    # After waking, try the queue again
+                    try:
+                        frame = q.get_nowait()
+                    except queue_mod.Empty:
+                        stream_removed = get_stream(job_id) is None
+                        continue
+
+                # Pace output at fixed 24fps
+                now = time.perf_counter()
+                wait = FRAME_INTERVAL - (now - last_yield_time)
+                if wait > 0:
+                    await asyncio.sleep(wait)
 
                 # Encode in thread (frame already resized by publish_frame)
                 encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 60]
                 success, buffer = await loop.run_in_executor(None, cv2.imencode, '.jpg', frame, encode_param)
 
                 if success:
+                    last_yield_time = time.perf_counter()
                     if not _first_yielded:
                         _first_yielded = True
                         if _stream_t0:
@@ -615,9 +638,6 @@ async def stream_video(job_id: str):
                     yield (b'--frame\r\n'
                            b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')
 
-                # Simple pacer (~30fps)
-                await asyncio.sleep(0.033)
-
             except Exception:
                 await asyncio.sleep(0.1)
 

inference.py

@@ -916,14 +916,11 @@ def run_inference(
                         writer.write(p_frame)
 
                         if stream_queue:
-                            try:
-                                from jobs.streaming import publish_frame as _publish
-                                if job_id:
-                                    _publish(job_id, p_frame)
-                                else:
-                                    stream_queue.put(p_frame, timeout=0.01)
-                            except:
-                                pass
+                            from jobs.streaming import publish_frame as _publish
+                            if job_id:
+                                _publish(job_id, p_frame)
+                            else:
+                                stream_queue.put(p_frame)
 
                         all_detections_map[next_idx] = dets
 
@@ -1244,7 +1241,7 @@ def run_grounded_sam2_tracking(
             return dets
 
         # Shared streaming state (publisher ↔ writer)
-        _stream_deque: collections.deque = collections.deque(maxlen=200)
+        _stream_deque: collections.deque = collections.deque()  # unbounded — publisher drains at its own pace
         _stream_lock = RLock()
         _stream_writer_done = Event()
 
@@ -1337,70 +1334,25 @@ def run_grounded_sam2_tracking(
                 _stream_writer_done.set()
 
         def _stream_publisher_thread():
-            """Adaptive-rate publisher: reads from _stream_deque, publishes at measured pace."""
+            """Forward every frame from deque to stream queue. Consumer handles pacing."""
             from jobs.streaming import publish_frame as _pub
 
-            STARTUP_WAIT = 5.0      # max seconds to accumulate before streaming
-            MIN_FPS = 2.0
-            MAX_FPS = 30.0
-            HEARTBEAT_INTERVAL = 0.5  # re-publish last frame if deque empty
-            LOW_WATER = 10
-            HIGH_WATER = 50
-            ADJUST_INTERVAL = 1.0
-
-            last_frame = None
             published = 0
 
-            # --- Phase 1: startup accumulation ---
-            t_start = time.perf_counter()
             while True:
-                elapsed = time.perf_counter() - t_start
-                if elapsed >= STARTUP_WAIT:
-                    break
-                if _stream_writer_done.is_set():
-                    break
-                time.sleep(0.1)
-
-            with _stream_lock:
-                accumulated = len(_stream_deque)
-            elapsed = max(time.perf_counter() - t_start, 0.1)
-            r_prod = accumulated / elapsed if accumulated > 0 else 10.0
-            r_stream = max(MIN_FPS, min(MAX_FPS, 0.85 * r_prod))
-
-            logging.info(
-                "Stream publisher started: R_prod=%.1f fps, R_stream=%.1f fps, "
-                "accumulated=%d frames in %.1fs",
-                r_prod, r_stream, accumulated, elapsed,
-            )
-            _ttfs(f"publisher: startup_wait done ({accumulated} frames in {elapsed:.1f}s)")
-
-            # --- Phase 2: adaptive streaming ---
-            last_adjust = time.perf_counter()
-            last_publish_time = 0.0
-
-            while True:
-                frame_interval = 1.0 / r_stream
-
-                # Try to pop a frame
                 frame = None
                 with _stream_lock:
                     if _stream_deque:
                         frame = _stream_deque.popleft()
 
                 if frame is not None:
-                    last_frame = frame
                     if job_id:
                         _pub(job_id, frame)
                     elif stream_queue:
-                        try:
-                            stream_queue.put(frame, timeout=0.01)
-                        except Exception:
-                            pass
+                        stream_queue.put(frame)
                     if published == 0:
                         _ttfs("first_publish_frame")
                     published += 1
-                    last_publish_time = time.perf_counter()
-                    time.sleep(frame_interval)
                 else:
                     # Deque empty — check termination
                     if _stream_writer_done.is_set():
@@ -1408,40 +1360,7 @@ def run_grounded_sam2_tracking(
                             if not _stream_deque:
                                 break
                         continue
-
-                    # Heartbeat: re-publish last frame to keep MJPEG alive
-                    now = time.perf_counter()
-                    if last_frame is not None and (now - last_publish_time) >= HEARTBEAT_INTERVAL:
-                        if job_id:
-                            _pub(job_id, last_frame)
-                        elif stream_queue:
-                            try:
-                                stream_queue.put(last_frame, timeout=0.01)
-                            except Exception:
-                                pass
-                        last_publish_time = now
-                    time.sleep(0.05)
-
-                # Adaptive rate adjustment (every ~1s)
-                now = time.perf_counter()
-                if now - last_adjust >= ADJUST_INTERVAL:
-                    with _stream_lock:
-                        level = len(_stream_deque)
-                    if level < LOW_WATER:
-                        r_stream = max(MIN_FPS, r_stream * 0.9)
-                    elif level > HIGH_WATER:
-                        r_stream = min(MAX_FPS, r_stream * 1.05)
-                    last_adjust = now
-
-            # Publish final frame
-            if last_frame is not None:
-                if job_id:
-                    _pub(job_id, last_frame)
-                elif stream_queue:
-                    try:
-                        stream_queue.put(last_frame, timeout=0.01)
-                    except Exception:
-                        pass
+                    time.sleep(0.01)  # brief wait for next frame
 
             logging.info("Stream publisher finished: published %d frames", published)
 

jobs/streaming.py

@@ -27,7 +27,7 @@ def _resize_for_stream(frame: np.ndarray) -> np.ndarray:
 def create_stream(job_id: str) -> queue.Queue:
     """Create a new stream queue for a job. Returns the queue for backward compat."""
     with _LOCK:
-        q = queue.Queue(maxsize=120)
+        q = queue.Queue()  # unbounded — no frame drops
         # Create event — will be lazily bound to the correct event loop
         event = None  # Lazily created in get_stream_event
         _STREAMS[job_id] = (q, event)
@@ -67,7 +67,7 @@ def remove_stream(job_id: str) -> None:
 
 
 def publish_frame(job_id: str, frame: Any) -> None:
-    """Publish a pre-resized frame to a job's stream. Non-blocking drop if full.
+    """Publish a pre-resized frame to a job's stream. Never drops frames.
 
     Also sets the asyncio.Event to wake the stream consumer immediately.
     """
@@ -81,11 +81,7 @@ def publish_frame(job_id: str, frame: Any) -> None:
     # Pre-resize for streaming (avoids resize in async handler)
     resized = _resize_for_stream(frame)
 
-    try:
-        q.put_nowait(resized)
-    except queue.Full:
-        # Drop frame if consumer is too slow
-        pass
+    q.put(resized)  # blocking put — no frame loss
 
     # Wake the async consumer if waiting
     if event is not None: