diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py | 58 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py | 299 |
2 files changed, 162 insertions, 195 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py index 05ab9e7..a425c65 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py @@ -641,3 +641,61 @@ class StreamingMixin: reason, index, time.monotonic() - self._stream_started_at) log.info("Streamed %s: %d segments", entry.name, index) self._audit("stream_video", entry.name) + + def _do_stream_request(self, msg: dict) -> None: + sem = self._ctx.get("_transcode_sem") + log.info("stream: req file=%s credits=%s slots_free=%s prev=%s", + str(msg.get("file_id"))[:12], msg.get("credits"), + getattr(sem, "_value", "?"), + "alive" if (self._stream_task and + not self._stream_task.done()) else "none") + self._spawn(self._replace_stream(msg)) + + def _do_client_diag(self, msg: dict) -> None: + # 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", " ") + if msg.get("event"): + # Once per stream or per seek, not once per five seconds — + # and a seek nobody asked for looks exactly like a viewer + # dragging the scrubber from this side, so it has to be + # visible without turning DEBUG on. + log.info( + "stream: client %s target=%s t=%ss offset=%s ready=%s " + "duration=%s ranges=[%s]", + _f("event", 16), _f("target"), _f("t"), _f("offset"), + _f("ready"), _f("duration"), _f("ranges", 120)) + # 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. + else: + # `ahead` on its own cannot say whether a short buffer is + # the player's own gate holding or the network failing to + # keep up, and those two want opposite answers. `limit` is + # what the gate is set to for this film and `budget` the + # byte budget it was derived from, so the three read as one + # sentence. + log.debug( + "stream: client t=%ss ahead=%ss/%ss budget=%sMB " + "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("limit"), _f("budgetMB"), + _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) + + def _do_stream_stop(self) -> None: + age = (time.monotonic() - self._stream_started_at + if self._stream_started_at else -1) + log.info("stream: stop received %.1fs after start, %d segments sent", + age, self._stream_segments) + self._stop_stream() diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py index cb2db8f..ee2e3a1 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py @@ -2,7 +2,6 @@ of them.""" import logging -import time from meshbay_common.protocol import MNP @@ -13,6 +12,96 @@ log = logging.getLogger("meshbay_node.transport.webrtc_server") # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 +# After the handshake: which method answers each message type, and whether it +# runs as a task of this session (`_spawn`) or before the next message is read. +# That choice is made per type, deliberately; the golden master in the tests +# records it for every type. +SPAWNED, INLINE = True, False +_HANDLERS = { + MNP.INDEX_SYNC: ("_do_index_sync", INLINE), + # Spawned rather than answered inline: the reply waits for room + # on the channel, and blocking the message loop for that would + # stop everything else this peer is doing — including the + # uploads whose acks free the very buffer we are waiting on. + # Chunks are matched by file and index on the client, so + # answering out of order is safe. + MNP.FILE_REQUEST: ("_do_file_request", SPAWNED), + MNP.CHAT_MESSAGE: ("_do_chat_message", INLINE), + MNP.CHAT_HISTORY: ("_do_chat_history", INLINE), + MNP.LINK_PREVIEW_REQ: ("_do_link_preview_request", SPAWNED), + MNP.PING: ("_do_ping", INLINE), + MNP.TRANSFER_OPEN: ("_do_transfer_open", INLINE), + MNP.TRANSFER_CLOSE: ("_do_transfer_close", INLINE), + MNP.FILE_UPLOAD: ("_do_file_upload", SPAWNED), + MNP.DIR_CREATE: ("_do_dir_create", SPAWNED), + MNP.DIR_DELETE: ("_do_dir_delete", SPAWNED), + MNP.FILE_DELETE: ("_do_file_delete", INLINE), + MNP.ADMIN_RESPONSE: ("_do_admin_response", INLINE), + MNP.INVITE_CREATE: ("_do_invite_create", INLINE), + MNP.INVITE_LINK_CREATE: ("_do_invite_link_create", INLINE), + MNP.INVITE_CANCEL: ("_do_invite_cancel", INLINE), + MNP.MEMBER_REVOKE: ("_do_member_revoke", INLINE), + MNP.DEVICE_REQUEST: ("_do_device_request", SPAWNED), + MNP.DEVICE_LOOKUP: ("_do_device_lookup", SPAWNED), + MNP.DEVICE_ADD: ("_do_device_add", SPAWNED), + MNP.DEVICE_LIST: ("_do_device_list", SPAWNED), + MNP.DEVICE_REVOKE: ("_do_device_revoke", SPAWNED), + MNP.DEVICE_HELLO: ("_do_device_hello", SPAWNED), + MNP.APPS_ENABLED: ("_do_apps_enabled", INLINE), + MNP.TRANSFER_LIMITS: ("_do_transfer_limits", INLINE), + MNP.SET_SCAN_SETTINGS: ("_do_set_scan_settings", INLINE), + MNP.TMDB_CONFIG: ("_do_tmdb_config", INLINE), + MNP.TMDB_ENABLED: ("_do_tmdb_enabled", INLINE), + MNP.APP_DIRECTORIES: ("_do_app_directories", INLINE), + MNP.CHAT_DIRECTORY: ("_do_chat_directory", INLINE), + MNP.CHAT_LINK_PREVIEW: ("_do_chat_link_preview", INLINE), + MNP.SEARCH_LISTED: ("_do_search_listed", INLINE), + MNP.CHAT_EPOCH: ("_do_chat_epoch", INLINE), + MNP.CHAT_KEYS_REQ: ("_do_chat_keys_req", SPAWNED), + MNP.GROUP_ROSTER_REQ: ("_do_group_roster_req", SPAWNED), + MNP.MEDIA_META_REQ: ("_do_media_meta_request", SPAWNED), + MNP.SEASON_META_REQ: ("_do_season_meta_request", SPAWNED), + MNP.TMDB_SEARCH_REQ: ("_do_tmdb_search_request", SPAWNED), + MNP.TMDB_OVERRIDE: ("_do_tmdb_override", INLINE), + MNP.TMDB_REMATCH: ("_do_tmdb_rematch", INLINE), + MNP.MUSICBRAINZ_ENABLED: ("_do_musicbrainz_enabled", INLINE), + MNP.MUSIC_META_REQ: ("_do_music_meta_request", SPAWNED), + MNP.AUDIO_TRANSCODE_REQ: ("_do_audio_transcode_request", SPAWNED), + MNP.SUBTITLE_REQ: ("_do_subtitle_request", SPAWNED), + MNP.MEMBER_UNPIN: ("_do_member_unpin", INLINE), + MNP.GEK_ROTATE: ("_do_gek_rotate", INLINE), + MNP.NODE_STATUS: ("_do_node_status", SPAWNED), + MNP.ROOT_ADD: ("_do_root_add", INLINE), + MNP.ROOT_REMOVE: ("_do_root_remove", INLINE), + MNP.ROOT_UPDATE: ("_do_root_update", INLINE), + MNP.ROOT_EJECT: ("_do_root_eject", INLINE), + MNP.ROOT_PLUG: ("_do_root_plug", INLINE), + MNP.ROSTER_READ: ("_do_roster_read", SPAWNED), + MNP.DENYLIST_READ: ("_do_denylist_read", SPAWNED), + MNP.DENYLIST_CLEAR: ("_do_denylist_clear", SPAWNED), + MNP.GROUP_ATTACH: ("_do_group_attach", INLINE), + MNP.GROUP_DETACH: ("_do_group_detach", INLINE), + MNP.NODE_SETTINGS_SET: ("_do_node_settings_set", SPAWNED), + MNP.NODE_RELOAD: ("_do_node_reload", SPAWNED), + MNP.KEYPAIR_BUNDLE_STORE: ("_do_keypair_bundle_store", SPAWNED), + MNP.KEYPAIR_BUNDLE_DELETE: ("_do_keypair_bundle_delete", SPAWNED), + MNP.USER_BLOB_STORE: ("_do_user_blob_store", SPAWNED), + MNP.USER_BLOB_FETCH: ("_do_user_blob_fetch", SPAWNED), + MNP.USER_BLOB_LIST: ("_do_user_blob_list", SPAWNED), + MNP.USER_BLOB_DELETE: ("_do_user_blob_delete", SPAWNED), + MNP.STREAM_REQUEST: ("_do_stream_request", INLINE), + MNP.STREAM_MORE: ("_grant_stream_credit", INLINE), + 'client_diag': ("_do_client_diag", INLINE), + MNP.STREAM_STOP: ("_do_stream_stop", INLINE), +} +# The handlers that are called without the message. +_TAKES_NO_MESSAGE = frozenset({ + "_do_index_sync", + "_do_keypair_bundle_delete", + "_do_stream_stop", + "_do_user_blob_list", +}) + class DispatchMixin: def _dispatch_message(self, msg: dict) -> None: @@ -49,201 +138,21 @@ class DispatchMixin: self._spawn(self._do_join_request(msg)) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) - elif mtype == MNP.INDEX_SYNC: - self._do_index_sync() - elif mtype == MNP.FILE_REQUEST: - # Spawned rather than answered inline: the reply waits for room - # on the channel, and blocking the message loop for that would - # stop everything else this peer is doing — including the - # uploads whose acks free the very buffer we are waiting on. - # Chunks are matched by file and index on the client, so - # answering out of order is safe. - self._spawn(self._do_file_request(msg)) - elif mtype == MNP.CHAT_MESSAGE: - self._do_chat_message(msg) - elif mtype == MNP.CHAT_HISTORY: - self._do_chat_history(msg) - elif mtype == MNP.LINK_PREVIEW_REQ: - self._spawn(self._do_link_preview_request(msg)) - elif mtype == MNP.PING: - self._do_ping(msg) - elif mtype == MNP.TRANSFER_OPEN: - self._do_transfer_open(msg) - elif mtype == MNP.TRANSFER_CLOSE: - self._do_transfer_close(msg) - elif mtype == MNP.FILE_UPLOAD: - self._spawn(self._do_file_upload(msg)) - elif mtype == MNP.DIR_CREATE: - self._spawn(self._do_dir_create(msg)) - elif mtype == MNP.DIR_DELETE: - self._spawn(self._do_dir_delete(msg)) - elif mtype == MNP.FILE_DELETE: - self._do_file_delete(msg) - elif mtype == MNP.ADMIN_RESPONSE: - self._do_admin_response(msg) - elif mtype == MNP.INVITE_CREATE: - self._do_invite_create(msg) - elif mtype == MNP.INVITE_LINK_CREATE: - self._do_invite_link_create(msg) - elif mtype == MNP.INVITE_CANCEL: - self._do_invite_cancel(msg) - elif mtype == MNP.MEMBER_REVOKE: - self._do_member_revoke(msg) - elif mtype == MNP.DEVICE_REQUEST: - self._spawn(self._do_device_request(msg)) - elif mtype == MNP.DEVICE_LOOKUP: - self._spawn(self._do_device_lookup(msg)) - elif mtype == MNP.DEVICE_ADD: - self._spawn(self._do_device_add(msg)) - elif mtype == MNP.DEVICE_LIST: - self._spawn(self._do_device_list(msg)) - elif mtype == MNP.DEVICE_REVOKE: - self._spawn(self._do_device_revoke(msg)) - elif mtype == MNP.DEVICE_HELLO: - self._spawn(self._do_device_hello(msg)) - elif mtype == MNP.APPS_ENABLED: - self._do_apps_enabled(msg) - elif mtype == MNP.TRANSFER_LIMITS: - self._do_transfer_limits(msg) - elif mtype == MNP.SET_SCAN_SETTINGS: - self._do_set_scan_settings(msg) - elif mtype == MNP.TMDB_CONFIG: - self._do_tmdb_config(msg) - elif mtype == MNP.TMDB_ENABLED: - self._do_tmdb_enabled(msg) - elif mtype == MNP.APP_DIRECTORIES: - self._do_app_directories(msg) - elif mtype == MNP.CHAT_DIRECTORY: - self._do_chat_directory(msg) - elif mtype == MNP.CHAT_LINK_PREVIEW: - self._do_chat_link_preview(msg) - elif mtype == MNP.SEARCH_LISTED: - self._do_search_listed(msg) - elif mtype == MNP.CHAT_EPOCH: - self._do_chat_epoch(msg) - elif mtype == MNP.CHAT_KEYS_REQ: - self._spawn(self._do_chat_keys_req(msg)) - elif mtype == MNP.GROUP_ROSTER_REQ: - self._spawn(self._do_group_roster_req(msg)) - elif mtype == MNP.MEDIA_META_REQ: - self._spawn(self._do_media_meta_request(msg)) - elif mtype == MNP.SEASON_META_REQ: - self._spawn(self._do_season_meta_request(msg)) - elif mtype == MNP.TMDB_SEARCH_REQ: - self._spawn(self._do_tmdb_search_request(msg)) - elif mtype == MNP.TMDB_OVERRIDE: - self._do_tmdb_override(msg) - elif mtype == MNP.TMDB_REMATCH: - self._do_tmdb_rematch(msg) - elif mtype == MNP.MUSICBRAINZ_ENABLED: - self._do_musicbrainz_enabled(msg) - elif mtype == MNP.MUSIC_META_REQ: - self._spawn(self._do_music_meta_request(msg)) - elif mtype == MNP.AUDIO_TRANSCODE_REQ: - self._spawn(self._do_audio_transcode_request(msg)) - elif mtype == MNP.SUBTITLE_REQ: - self._spawn(self._do_subtitle_request(msg)) - elif mtype == MNP.MEMBER_UNPIN: - self._do_member_unpin(msg) - elif mtype == MNP.GEK_ROTATE: - self._do_gek_rotate(msg) - elif mtype == MNP.NODE_STATUS: - self._spawn(self._do_node_status(msg)) - elif mtype == MNP.ROOT_ADD: - self._do_root_add(msg) - elif mtype == MNP.ROOT_REMOVE: - self._do_root_remove(msg) - elif mtype == MNP.ROOT_UPDATE: - self._do_root_update(msg) - elif mtype == MNP.ROOT_EJECT: - self._do_root_eject(msg) - elif mtype == MNP.ROOT_PLUG: - self._do_root_plug(msg) - elif mtype == MNP.ROSTER_READ: - self._spawn(self._do_roster_read(msg)) - elif mtype == MNP.DENYLIST_READ: - self._spawn(self._do_denylist_read(msg)) - elif mtype == MNP.DENYLIST_CLEAR: - self._spawn(self._do_denylist_clear(msg)) - elif mtype == MNP.GROUP_ATTACH: - self._do_group_attach(msg) - elif mtype == MNP.GROUP_DETACH: - self._do_group_detach(msg) - elif mtype == MNP.NODE_SETTINGS_SET: - self._spawn(self._do_node_settings_set(msg)) - elif mtype == MNP.NODE_RELOAD: - self._spawn(self._do_node_reload(msg)) - elif mtype == MNP.KEYPAIR_BUNDLE_STORE: - self._spawn(self._do_keypair_bundle_store(msg)) - elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: - self._spawn(self._do_keypair_bundle_delete()) - elif mtype == MNP.USER_BLOB_STORE: - self._spawn(self._do_user_blob_store(msg)) - elif mtype == MNP.USER_BLOB_FETCH: - self._spawn(self._do_user_blob_fetch(msg)) - elif mtype == MNP.USER_BLOB_LIST: - self._spawn(self._do_user_blob_list()) - elif mtype == MNP.USER_BLOB_DELETE: - self._spawn(self._do_user_blob_delete(msg)) - elif mtype == MNP.STREAM_REQUEST: - sem = self._ctx.get("_transcode_sem") - log.info("stream: req file=%s credits=%s slots_free=%s prev=%s", - str(msg.get("file_id"))[:12], msg.get("credits"), - getattr(sem, "_value", "?"), - "alive" if (self._stream_task and - not self._stream_task.done()) else "none") - 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", " ") - if msg.get("event"): - # Once per stream or per seek, not once per five seconds — - # and a seek nobody asked for looks exactly like a viewer - # dragging the scrubber from this side, so it has to be - # visible without turning DEBUG on. - log.info( - "stream: client %s target=%s t=%ss offset=%s ready=%s " - "duration=%s ranges=[%s]", - _f("event", 16), _f("target"), _f("t"), _f("offset"), - _f("ready"), _f("duration"), _f("ranges", 120)) - # 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. - else: - # `ahead` on its own cannot say whether a short buffer is - # the player's own gate holding or the network failing to - # keep up, and those two want opposite answers. `limit` is - # what the gate is set to for this film and `budget` the - # byte budget it was derived from, so the three read as one - # sentence. - log.debug( - "stream: client t=%ss ahead=%ss/%ss budget=%sMB " - "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("limit"), _f("budgetMB"), - _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) - log.info("stream: stop received %.1fs after start, %d segments sent", - age, self._stream_segments) - self._stop_stream() else: - log.warning("Unknown MNP message type on DataChannel: %s", mtype) + try: + entry = _HANDLERS.get(mtype) + except TypeError: + # A list or a map is a legal msgpack value for `type` and not a + # key: it names no handler, as it matched no comparison before. + entry = None + if entry is None: + log.warning("Unknown MNP message type on DataChannel: %s", mtype) + else: + name, spawned = entry + method = getattr(self, name) + result = method() if name in _TAKES_NO_MESSAGE else method(msg) + if spawned: + self._spawn(result) except Exception as e: # Log the detail locally; send the peer a generic message. Exception # text here carries filesystem paths and internal state (finding L3). |