feat: visualize retained network playback buffer

Checks/streaming.py

@@ -14,8 +14,9 @@ from urllib.parse import parse_qs, unquote, urlsplit
 
 
 class _Gate:
-    def __init__(self, total: int):
+    def __init__(self, total: int, prefix_ends: dict[str, float]):
         self.total = total
+        self.prefix_ends = prefix_ends
         self.entered = threading.Event()
         self.release = threading.Event()
         self.claimed = False
@@ -45,6 +46,7 @@ class StreamingServer:
         self.lock = threading.Lock()
         self.gates: dict[str, _Gate] = {}
         self.stats: dict[str, dict[str, int]] = {}
+        self.prefixes: dict[tuple[str, int], dict[str, float]] = {}
         self.stopping = threading.Event()
         owner = self
 
@@ -103,6 +105,35 @@ class StreamingServer:
         end = min(int(last), size - 1) if last else size - 1
         return (start, end) if 0 <= start <= end < size else None
 
+    def _prefix_ends(self, name: str, offset: int) -> dict[str, float]:
+        """Timestamp upper bounds, never a byte fraction interpreted as playback time."""
+        key = (name, offset)
+        if key not in self.prefixes:
+            probe = subprocess.run(
+                ["ffprobe", "-v", "error", "-show_packets", "-show_streams", "-show_format",
+                 "-show_entries", "format=start_time:stream=index,codec_type:packet=stream_index,pts_time,duration_time,pos,size",
+                 "-of", "json", str(self.fixtures / name)],
+                capture_output=True, text=True, check=True, timeout=30,
+            )
+            metadata = json.loads(probe.stdout)
+            tracks = {stream["index"]: stream["codec_type"] for stream in metadata["streams"]}
+            origin = float(metadata.get("format", {}).get("start_time", 0))
+            ends = {"audio": 0.0, "video": 0.0}
+            first_missing = {"audio": float("inf"), "video": float("inf")}
+            for packet in metadata["packets"]:
+                kind = tracks.get(packet["stream_index"])
+                if kind not in ends or "pts_time" not in packet or "pos" not in packet:
+                    continue
+                # A packet crossing the gate is not playable, even if its header arrived.
+                if int(packet["pos"]) + int(packet["size"]) <= offset:
+                    end = float(packet["pts_time"]) - origin + float(packet.get("duration_time", 0))
+                    ends[kind] = max(ends[kind], end)
+                else:
+                    first_missing[kind] = min(first_missing[kind], max(0, float(packet["pts_time"]) - origin))
+            # A reordered reference packet's later PTS cannot bridge an undelivered B-frame.
+            self.prefixes[key] = {kind: min(end, first_missing[kind]) for kind, end in ends.items()}
+        return self.prefixes[key]
+
     def _get(self, request):
         parsed = urlsplit(request.path)
         parts = unquote(parsed.path).strip("/").split("/")
@@ -116,13 +147,13 @@ class StreamingServer:
                 gate = self.gates.get(parts[2])
                 if gate is None:
                     body = {"entered": False, "released": False, "waiting": 0, "requests": 0,
-                            "ranges": 0, "bytes": 0, "total": 0}
+                            "ranges": 0, "bytes": 0, "total": 0, "prefix_ends": {}}
                 else:
                     if parts[1] == "release":
                         gate.release.set()
                     body = {"entered": gate.entered.is_set(), "released": gate.release.is_set(),
                             "waiting": gate.waiting, "requests": gate.requests, "ranges": gate.ranges,
-                            "bytes": gate.bytes, "total": gate.total}
+                            "bytes": gate.bytes, "total": gate.total, "prefix_ends": gate.prefix_ends}
             self._reply(request, 200, json.dumps(body).encode())
             return
 
@@ -180,7 +211,10 @@ class StreamingServer:
                 self._reply(request, 400, b"Invalid gate offset", "text/plain")
                 return
             with self.lock:
-                gate = self.gates.setdefault(token, _Gate(size))
+                if token not in self.gates:
+                    prefix = self._prefix_ends(name, gate_at) if mode == "stall" else {}
+                    self.gates[token] = _Gate(size, prefix)
+                gate = self.gates[token]
                 gate.requests += 1
                 gate.ranges += int(range_header is not None)
                 # Only the original prefix connection is gated: a seek can create a new