diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 74 |
1 files changed, 41 insertions, 33 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 d79674d..e748be9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -139,7 +139,8 @@ from meshbay_node.media_probe import ( probe_video as _probe_video, ) from meshbay_node.roots import ( - ROOT_NOT_SERVED, RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, + ROOT_NOT_SERVED, RootSet, entry_abs_path, off_disk, SAFE_UPLOAD_NAME, safe_subdir, + _free_name, ) log = logging.getLogger(__name__) @@ -3877,12 +3878,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return # A real index entry, asked for without a lease: browsing, or a client @@ -3913,7 +3911,8 @@ class WebRTCPeerSession: file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) - chunk_data = _read_and_encrypt( + chunk_data = await off_disk( + ctx["roots"], _read_and_encrypt, ctx["gek"], file_path, chunk_index, file_hash, entry.id) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for @@ -4007,12 +4006,9 @@ class WebRTCPeerSession: if not entry: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return media_cache = self._ctx.get("media_cache") @@ -4074,12 +4070,9 @@ class WebRTCPeerSession: if not entry: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return media_cache = self._ctx.get("media_cache") @@ -5882,7 +5875,7 @@ class WebRTCPeerSession: self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") return - self._exec_file_delete(ctx, file_id, entry) + await self._exec_file_delete(ctx, file_id, entry) async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, @@ -5918,15 +5911,15 @@ class WebRTCPeerSession: "username": result.get("username", ""), }) - def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: + async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal == ROOT_NOT_SERVED: # Frozen, not gone: removing the entry would lose a file that is # still on a drive the node cannot read right now. self._send({"type": "error", "detail": ROOT_NOT_SERVED}) return - if file_path.exists(): - file_path.unlink() + if file_path is not None: + await off_disk(ctx["roots"], file_path.unlink) log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) @@ -6123,12 +6116,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return gek = ctx.get("gek") @@ -6581,6 +6571,24 @@ class WebRTCPeerSession: await self._pc.close() +def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: + """ + Where an entry is, and whether it is readable — or the refusal to send. + + Both halves are syscalls: `resolve()` walks the path and `exists()` stats + it, and a stat is what *wakes* a sleeping disk. Leaving either on the event + loop and offloading only the read would move the stall rather than remove + it, and the read would then find the disk already awake. Blocking; called + through `off_disk`. + """ + path = entry_abs_path(roots, entry) + if path is None: + return None, ROOT_NOT_SERVED + if not path.exists(): + return None, "File not on disk" + return path, None + + def _read_and_encrypt( gek: bytes, file_path: Path, @@ -6588,7 +6596,7 @@ def _read_and_encrypt( file_hash: bytes, file_id: str = "", ) -> dict: - """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short.""" + """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) |