aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-15 19:01:08 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-15 19:01:08 +0200
commit05f4feab641740c944d636f29a03f8c0dd1328c7 (patch)
tree9a41cbaac3c36db854d6a1eb7750404ff79fa4e4 /packages/meshbay-node/src
parentdd3927a661273734493f65a593755b95aecf5f09 (diff)
downloadmeshbay-05f4feab641740c944d636f29a03f8c0dd1328c7.tar.gz
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>
Diffstat (limited to 'packages/meshbay-node/src')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py41
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,