fix: stop a stream on close, count only real users, record where a node is

**Closing the viewer left the node working.** Nothing told it to stop:
the player dropped its handlers, which only made the browser deaf. ffmpeg
kept running and held one of the node's two transcode slots until the
credit timeout expired two minutes later — which is why the next video
answered "server busy". `stream_stop` ends it at once, and the viewer
also drops its queue, ends the MediaSource and revokes the object URL on
the way out, any of which could be holding megabytes of decrypted video.

While there: `file_chunk` replies were matched to their requests by
arrival order, which was true by luck rather than by construction. The
reply now names the file it belongs to and is matched on that and the
chunk index; a chunk nobody is waiting for is dropped instead of being
handed to whatever request happens to be oldest.

**The administration panel counted its own history.** A deleted account
is tombstoned so the connection log stays readable, and every count and
list treated that row as a user — including a group's member count, and
the member list of the group itself. They do not any more.

**Where a node is.** `endpoint_hint` is what a node believes its address
to be, learned from a STUN server and sent to us: useful for reaching it,
and a claim. The announcement that carries it is signed with the node key
over a fresh timestamp, so the address that request *arrives from* is the
address of whoever holds that key — that is now recorded on the node row
and shown in a Nodes tab, next to the hint, with the difference spelled
out. Clients get the same treatment: `webrtc_offer` is logged with the
address the hub saw when a browser starts a peer connection.

Verified against the live deployment: the node's row reads 90.112.206.172
after a restart, and in e2e a stopped stream goes quiet in one message
and the next one starts immediately instead of being refused.

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

1 files changed, 36 insertions, 5 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 eab1cac..a892be2 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -297,6 +297,7 @@ class WebRTCPeerSession:
         # Flow control for video: how many segments the client says it can take.
         self._stream_credit = 0
         self._stream_credit_evt = asyncio.Event()
+        self._stream_stopped = False
         self._gek_challenge: bytes | None = None
         # Same value as the GEK challenge, but kept for the life of the connection:
         # a join_request is signed over it, and it must stay verifiable after the
@@ -384,6 +385,8 @@ class WebRTCPeerSession:
                 asyncio.ensure_future(self._stream_video(msg))
             elif mtype == MNP.STREAM_MORE:
                 self._grant_stream_credit(msg)
+            elif mtype == MNP.STREAM_STOP:
+                self._stop_stream()
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
         except Exception as e:
@@ -1203,6 +1206,7 @@ class WebRTCPeerSession:
             file_path,
             chunk_index,
             file_hash,
+            entry.id,
         )
         self._send(chunk_data)
         if chunk_index == 0:
@@ -1721,6 +1725,18 @@ class WebRTCPeerSession:
         self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT))
         self._stream_credit_evt.set()
 
+    def _stop_stream(self) -> None:
+        """
+        The viewer was closed. Stop transcoding and let go of the slot.
+
+        Without this the only thing that ended a stream was the credit timeout,
+        so ffmpeg kept running and held one of the node's two transcode slots
+        for two minutes after nobody was watching — which is how closing a video
+        made the next one answer "server busy".
+        """
+        self._stream_stopped = True
+        self._stream_credit_evt.set()
+
     async def _await_stream_credit(self) -> bool:
         """
         Block until the client has room. False if it stopped asking.
@@ -1730,6 +1746,8 @@ class WebRTCPeerSession:
         JavaScript array while MediaSource consumes it a segment at a time.
         """
         while self._stream_credit <= 0:
+            if self._stream_stopped:
+                return False
             self._stream_credit_evt.clear()
             try:
                 await asyncio.wait_for(self._stream_credit_evt.wait(),
@@ -1738,6 +1756,8 @@ class WebRTCPeerSession:
                 log.info("Stream stalled: no credit from peer=%s",
                          (self._user_id or "?")[:8])
                 return False
+            if self._stream_stopped:
+                return False
             if self._channel is None or self._channel.readyState != "open":
                 return False
         self._stream_credit -= 1
@@ -1809,12 +1829,17 @@ class WebRTCPeerSession:
         except (TypeError, ValueError):
             self._stream_credit = 0
         paced = self._stream_credit > 0
+        self._stream_stopped = False
 
         index = 0
         try:
             while True:
                 if paced and not await self._await_stream_credit():
                     break
+                if self._stream_stopped:
+                    log.info("Stream stopped by peer=%s after %d segments",
+                             (self._user_id or "?")[:8], index)
+                    break
                 data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
                 if not data:
                     break
@@ -1840,11 +1865,12 @@ class WebRTCPeerSession:
                 pass
             await proc.wait()
 
-        self._send({
-            "type": MNP.STREAM_END,
-            "v": MNP_VERSION,
-            "file_id": file_id,
-        })
+        if not self._stream_stopped:
+            self._send({
+                "type": MNP.STREAM_END,
+                "v": MNP_VERSION,
+                "file_id": file_id,
+            })
         log.info("Streamed %s: %d segments", entry.name, index)
         self._audit("stream_video", entry.name)
 
@@ -1868,6 +1894,7 @@ def _read_and_encrypt(
     file_path: Path,
     chunk_index: int,
     file_hash: bytes,
+    file_id: str = "",
 ) -> dict:
     with open(file_path, "rb") as f:
         f.seek(chunk_index * CHUNK_SIZE)
@@ -1879,6 +1906,10 @@ def _read_and_encrypt(
     return {
         "type": MNP.FILE_CHUNK,
         "v": MNP_VERSION,
+        # Named so a client running several downloads at once can tell whose
+        # reply this is. It used to carry only the index, which made matching a
+        # reply to its request a question of arrival order.
+        "file_id": file_id,
         "chunk_index": chunk_index,
         "plaintext_size": len(plaintext),
         "nonce": nonce,