feat(node): the stream capacity is the operator's to set, and a log that says who stopped

Bounding the client's read-ahead changed what a transcode slot is. It used to
be a burst — the browser took segments as fast as it could append them, so a
slot came back within the minute whatever the length of the film. Now it is
held for as long as someone is watching, so the cap counts simultaneous
viewers, and two of them meant the third was refused for the next hour and a
half.

The right number depends on the machine, so it belongs to the operator:
`[node] max_concurrent_streams` in node.toml, or
MESHBAY_MAX_CONCURRENT_STREAMS. Default 8 — one ffmpeg per viewer, remuxing
rather than encoding, idle on a pipe for most of the film. Zero, a negative
number, a non-number and a bool are refused with a warning naming the setting:
`Semaphore(0)` is not "no limit", it is a node where no video ever plays and
nothing says why, and TOML `true` would have become 1 by way of `int()`.

A stream also ends on the peer's silence now rather than on its stinginess. A
viewer buffered well ahead deliberately grants nothing for minutes, and the
old budget accumulated over the whole wait, so a keepalive that granted no
credit could not keep a paused film alive.

The rest is diagnosis, which is what this cost. `client_diag` carries the
player's own view — readyState, refused appends, buffered ranges, the video
element's error — into the node's log at DEBUG, next to the node's view of the
same stream. It is the only window into a phone, and every field is
stringified and cut short because all of it is peer-controlled. The node also
logs the first keepalive, which distinguishes a paced client from an unpaced
one at a glance, and progress every hundred segments, whose last line says
where a stream stopped and which side stopped it.

1 files changed, 115 insertions, 17 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 91d801c..5f5c9ff 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -90,7 +90,20 @@ MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file
 PRE_HANDSHAKE_MAX_MSG = 64 * 1024
 # ffmpeg is spawned per stream request; without a cap any member can fork-bomb
 # the node by requesting many streams at once (H6).
-MAX_CONCURRENT_TRANSCODES = 2
+#
+# Two was sized when a stream was a burst: the client took segments as fast as
+# it could append them, so a slot was held for the minute it took to push the
+# file and then came back. Now that the client only pulls ninety seconds ahead
+# of the playhead, a slot is held for as long as the film runs — so two slots
+# means two people can watch anything at all, and the third is refused for the
+# next hour and a half. The work behind a slot has not changed and is small:
+# ffmpeg runs `-c copy`, a remux with no encoding in it, and spends most of the
+# film blocked on a pipe nobody is reading.
+#
+# This is the default, not the policy: the right number depends on the machine,
+# so the operator sets `max_concurrent_streams` under [node] in node.toml. This
+# value applies when they have said nothing.
+MAX_CONCURRENT_TRANSCODES = 8
 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited
 # until the native client removes remote keypair bundles entirely.
 MAX_PRE_PROOF_FETCHES = 4
@@ -326,6 +339,12 @@ class WebRTCPeerSession:
         self._stream_credit = 0
         self._stream_credit_evt = asyncio.Event()
         self._stream_stopped = False
+        # When the peer last said anything about this stream. See
+        # _await_stream_credit: silence is what ends a stream, not stinginess.
+        self._stream_heard_at = 0.0
+        # Diagnostics: how many `stream_more n=0` the peer sent. See
+        # _grant_stream_credit — it tells a paced client from an unpaced one.
+        self._stream_keepalives = 0
         # The stream this session currently owns. One viewer plays one film at
         # a time, so a second request means the first is over — see
         # _replace_stream for why waiting for it to time out is not an option.
@@ -434,6 +453,28 @@ class WebRTCPeerSession:
                 self._spawn(self._replace_stream(msg))
             elif mtype == MNP.STREAM_MORE:
                 self._grant_stream_credit(msg)
+            elif mtype == "client_diag":
+                # Diagnostics only. The node acts on none of it — it writes it
+                # next to its own view of the same stream, which is the only
+                # place the two halves can be compared when the client is a
+                # phone with no console.
+                # Every field is peer-controlled, so each is stringified and
+                # cut short: this is a log line, not a channel for writing
+                # whatever one likes into the operator's file.
+                def _f(key: str, n: int = 24) -> str:
+                    return str(msg.get(key))[:n].replace("\n", " ")
+                # Debug: one line every five seconds per viewer. Run the daemon
+                # with --log-level debug to see inside a player that is
+                # misbehaving — it is the only view of the browser there is
+                # when the browser is a phone.
+                log.debug(
+                    "stream: client t=%ss ahead=%ss ready=%s paused=%s "
+                    "stalled=%s q=%s inflight=%s appending=%s updating=%s "
+                    "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)",
+                    _f("t"), _f("ahead"), _f("ready"), _f("paused"),
+                    _f("stalled"), _f("q"), _f("inflight"), _f("appending"),
+                    _f("updating"), _f("quota"), _f("ms"), _f("err", 80),
+                    _f("ranges", 120), self._stream_segments)
             elif mtype == MNP.STREAM_STOP:
                 age = (time.monotonic() - self._stream_started_at
                        if self._stream_started_at else -1)
@@ -1325,10 +1366,7 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not on disk"})
             return
 
-        sem = self._ctx.get("_transcode_sem")
-        if sem is None:
-            sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES)
-            self._ctx["_transcode_sem"] = sem
+        sem = self._transcode_semaphore()
 
         try:
             async with sem:
@@ -1830,14 +1868,30 @@ class WebRTCPeerSession:
         })
 
     def _grant_stream_credit(self, msg: dict) -> None:
+        """
+        The client has room for more segments.
+
+        `n` of zero is a keepalive, not a no-op: a viewer whose buffer is
+        already a minute and a half ahead of the playhead deliberately grants
+        nothing, and must still be able to say it is there. Without that, the
+        stall timeout below cannot tell a paused film from a closed tab.
+        """
         log.debug("stream credit +%s (had %d, sent %d)",
                   msg.get("n"), self._stream_credit, self._stream_segments)
-        """The client has room for more segments."""
         try:
             n = int(msg.get("n", 1))
         except (TypeError, ValueError):
             n = 1
+        if n == 0:
+            # The fingerprint of a client that bounds its read-ahead. A client
+            # that never sends one is granting credit per append — which is
+            # what fills the browser's buffer ceiling and wedges the player.
+            self._stream_keepalives += 1
+            if self._stream_keepalives == 1:
+                log.info("stream: peer is pacing itself (first keepalive at "
+                         "%d segments)", self._stream_segments)
         self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT))
+        self._stream_heard_at = time.monotonic()
         self._stream_credit_evt.set()
 
     def _stop_stream(self) -> None:
@@ -1860,8 +1914,22 @@ class WebRTCPeerSession:
         fast as it is produced, and the browser holds a four gigabyte film in a
         JavaScript array while MediaSource consumes it a segment at a time.
         """
-        waited = 0.0
+        # Measured from the last thing the peer said, not from the start of the
+        # wait: a viewer that is buffered well ahead sends keepalives and grants
+        # nothing for minutes at a time, and that is a watched film, not a
+        # stalled one.
+        self._stream_heard_at = time.monotonic()
+        waiting_since = 0.0
         while self._stream_credit <= 0:
+            if waiting_since == 0.0:
+                waiting_since = time.monotonic()
+                # Debug: a paced viewer runs out of credit between every
+                # window, so this is one line per eight segments — hundreds
+                # per film. It is worth having, but not by default.
+                log.debug("stream: out of credit at %d segments (%.0f MB) — "
+                          "waiting for the peer",
+                          self._stream_segments,
+                          self._stream_segments * STREAM_SEGMENT_SIZE / 1048576)
             if self._stream_stopped:
                 return False
             # Checked before the wait as well as after it: a peer that vanishes
@@ -1877,16 +1945,23 @@ class WebRTCPeerSession:
                 await asyncio.wait_for(self._stream_credit_evt.wait(),
                                        timeout=STREAM_CREDIT_POLL)
             except asyncio.TimeoutError:
-                waited += STREAM_CREDIT_POLL
-                if waited >= STREAM_CREDIT_TIMEOUT:
-                    log.info("Stream stalled: no credit from peer=%s",
-                             (self._user_id or "?")[:8])
+                silent = time.monotonic() - self._stream_heard_at
+                if silent >= STREAM_CREDIT_TIMEOUT:
+                    log.info("Stream stalled: nothing from peer=%s for %.0fs",
+                             (self._user_id or "?")[:8], silent)
                     return False
                 continue
             if self._stream_stopped:
                 return False
             if self._channel is None or self._channel.readyState != "open":
                 return False
+        if waiting_since:
+            waited_for = time.monotonic() - waiting_since
+            # Only a wait long enough to be a symptom. Normal pacing puts a
+            # gap of a few seconds between windows; a minute means the viewer
+            # is buffered right up and playing, or has stopped watching.
+            level = log.info if waited_for >= 10 else log.debug
+            level("stream: credit arrived after %.1fs", waited_for)
         self._stream_credit -= 1
         return True
 
@@ -1924,15 +1999,26 @@ class WebRTCPeerSession:
         self._stream_task = asyncio.current_task()
         await self._stream_video(msg)
 
-    async def _stream_video(self, msg: dict) -> None:
-        """Stream a video file as fMP4 segments via MSE-compatible output."""
-        # One ffmpeg per request with no cap lets any member exhaust the node's
-        # CPU and process table (H6). The semaphore lives on the transport context
-        # so it is shared across all peers, not per-session.
+    def _transcode_semaphore(self) -> asyncio.Semaphore:
+        """The node's stream budget, shared across every peer.
+
+        One ffmpeg per request with no cap lets any member exhaust the node's
+        CPU and process table (H6). The semaphore lives on the transport
+        context rather than the session so that it counts the node's viewers
+        and not one browser's, and it is created once: rebuilding it per call
+        would hand every caller its own budget and cap nothing at all.
+        """
         sem = self._ctx.get("_transcode_sem")
         if sem is None:
-            sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES)
+            n = self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES
+            sem = asyncio.Semaphore(n)
             self._ctx["_transcode_sem"] = sem
+            log.info("stream: %d concurrent viewers allowed", n)
+        return sem
+
+    async def _stream_video(self, msg: dict) -> None:
+        """Stream a video file as fMP4 segments via MSE-compatible output."""
+        sem = self._transcode_semaphore()
         if sem.locked() and sem._value <= 0:
             self._send({"type": "error", "detail": "Server busy, retry shortly"})
             return
@@ -2029,6 +2115,14 @@ class WebRTCPeerSession:
                 })
                 index += 1
                 self._stream_segments = index
+                if index % 100 == 0:
+                    # A stream that stops shows up here as a last line, and the
+                    # numbers on it say which side stopped it.
+                    log.info("stream: %d segments (%.0f MB), credit=%d, "
+                             "keepalives=%d, %.0fs in",
+                             index, index * STREAM_SEGMENT_SIZE / 1048576,
+                             self._stream_credit, self._stream_keepalives,
+                             time.monotonic() - self._stream_started_at)
                 await asyncio.sleep(0)
         except Exception as e:
             log.error("Stream error: %s", e)
@@ -2150,6 +2244,7 @@ class WebRTCTransport:
         groups: dict[str, dict] | None = None,
         denylist: Any | None = None,
         stun_servers: list[str] | None = None,
+        max_concurrent_streams: int | None = None,
     ):
         self._ctx: dict[str, Any] = {
             "sk_node": sk_node,
@@ -2158,6 +2253,9 @@ class WebRTCTransport:
             "shared_root": shared_root,
             "index": index,
             "_peers": {},
+            # None means "the operator said nothing" — the default applies. It
+            # is read once, when the first stream builds the semaphore.
+            "max_concurrent_streams": max_concurrent_streams,
         }
         if groups:
             self._ctx["groups"] = groups