diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_disk_io_off_loop.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_disk_io_off_loop.py | 224 |
1 files changed, 224 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py new file mode 100644 index 0000000..206b77b --- /dev/null +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -0,0 +1,224 @@ +""" +A slow disk must cost the caller that touched it, and nobody else. + +A root that has spun down, or that lives on a network mount, answers its first +syscall in seconds rather than microseconds. Made from the event loop, that +stalls the entire node: no other group is served, no stream is fed, no chat +message is delivered, and the hub socket is not read, for as long as the platter +takes to come back. It was found from the other end — a client's connection +attempt timing out on a node with one member, while the disk woke up. + +These tests do not read the source to check which thread a call is made from. +They make the disk slow and **measure whether the loop kept running**: a ticker +counts its own wake-ups beside the request, and a handler that blocks the loop +takes every one of those wake-ups with it. Put the call back inline and the +ticker count collapses to zero, which is the property being guarded. + +The third test is the one that is not about latency: one worker per root set +means two reads of the same file can never be inside it at once, and that is +what makes the single `f.seek()`/`f.read()` pair safe without a lock. +""" + +import asyncio +import re +import threading +import time +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_common.protocol import MNP +from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.transfers import LeaselessReads +from meshbay_node.transport import webrtc_server +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +GROUP = "g" * 32 + +# Long enough that a blocked loop is unmistakable, short enough to keep the +# suite quick. The ticker below wakes every 5 ms, so a loop that stays free +# gets ~40 wake-ups inside one of these and a blocked one gets none. +SLOW_S = 0.2 +TICK_S = 0.005 +CONTENT = b"a file worth waking a disk for" * 400 + + +class _Channel: + readyState = "open" + bufferedAmount = 0 + + +class _Session(WebRTCPeerSession): + def __init__(self, ctx): + self._ctx = ctx + self._registry_key = "s1" + self._user_id = "alice" + self._username = "alice" + self._group_id = GROUP + self._channel = _Channel() + self._leaseless = LeaselessReads() + self._unleased_noted = False + self.sent: list[dict] = [] + + def _send(self, msg): + self.sent.append(msg) + + def _audit(self, event, detail=""): + pass + + def _spawn(self, coro): + coro.close() + return None + + +async def _served(tmp_path: Path): + """A session serving one real file out of one real root.""" + root = tmp_path / "films" + root.mkdir() + (root / "clip.bin").write_bytes(CONTENT) + roots = RootSet.build([{"path": str(root), "name": "films"}]) + gek = generate_gek() + idx = DirectoryIndexer(roots=roots, group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await idx.initial_scan() + entry = next(e for e in idx.index.entries if e.name == "clip.bin") + ctx = {"_peers": {}, "roots": roots, "index": idx.index, + "sk_node": idx.sk_node, "gek": gek} + return _Session(ctx), ctx, entry, gek + + +class _Ticker: + """Counts how many times the event loop came back to it.""" + + def __init__(self): + self.ticks = 0 + self._stop = False + self._task: asyncio.Task | None = None + + async def _run(self): + while not self._stop: + await asyncio.sleep(TICK_S) + self.ticks += 1 + + def __enter__(self): + self._task = asyncio.get_running_loop().create_task(self._run()) + return self + + def __exit__(self, *exc): + self._stop = True + self._task.cancel() + return False + + +def _slow(fn): + """`fn`, plus a sleep on whichever thread calls it.""" + + def wrapper(*args, **kwargs): + time.sleep(SLOW_S) + return fn(*args, **kwargs) + + return wrapper + + +async def test_a_slow_chunk_read_does_not_stop_the_loop(tmp_path, monkeypatch): + session, _, entry, gek = await _served(tmp_path) + monkeypatch.setattr(webrtc_server, "_read_and_encrypt", + _slow(webrtc_server._read_and_encrypt)) + + with _Ticker() as ticker: + await session._do_file_request( + {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0}) + + # The read really did take its time, and the loop really did keep running: + # both halves matter, because a wrapper that never ran would also leave the + # ticker free. + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s read") + + chunk = next(m for m in session.sent if m.get("type") == MNP.FILE_CHUNK) + key = chunk_key_aes(gek, bytes.fromhex(entry.id), 0) + assert decrypt_chunk_aes(key, chunk["nonce"], chunk["ct"]) == CONTENT + + +async def test_a_slow_stat_does_not_stop_the_loop(tmp_path, monkeypatch): + """ + The stat matters as much as the read: it is what *wakes* the disk. + + Offloading only the read would leave the spin-up on the loop and the read + would then find the disk already awake — the stall moved, not removed. + """ + session, _, entry, _ = await _served(tmp_path) + monkeypatch.setattr(webrtc_server, "_locate", _slow(webrtc_server._locate)) + + with _Ticker() as ticker: + await session._do_file_request( + {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0}) + + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s stat") + assert any(m.get("type") == MNP.FILE_CHUNK for m in session.sent) + + +async def test_one_root_set_reads_one_chunk_at_a_time(tmp_path): + """ + Two reads of the same root are never in flight together. + + Not a performance choice: interleaved reads of one spinning drive seek-thrash + (the indexer's executor carries the measurement), and a second thread inside + the same `open`/`seek`/`read` sequence is a correctness question this avoids + having to answer. A pool would reopen both. + """ + session, _, entry, _ = await _served(tmp_path) + inside = 0 + peak = 0 + guard = threading.Lock() + real = webrtc_server._read_and_encrypt + + def counting(*args, **kwargs): + nonlocal inside, peak + with guard: + inside += 1 + peak = max(peak, inside) + try: + time.sleep(0.02) + return real(*args, **kwargs) + finally: + with guard: + inside -= 1 + + webrtc_server._read_and_encrypt = counting + try: + await asyncio.gather(*[ + session._do_file_request( + {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0}) + for _ in range(6) + ]) + finally: + webrtc_server._read_and_encrypt = real + + assert peak == 1, f"{peak} reads of one root set were inside the disk at once" + + +def test_no_handler_resolves_a_path_on_the_loop(): + """ + `entry_abs_path` is `Path.resolve()`, which is syscalls — it belongs on the + disk thread with everything else. + + This reads the source because there is nothing else to read: a handler added + later that resolves an entry inline would pass every test above, since those + only exercise the handlers that exist today. `_locate` is the one place + allowed to call it, and `off_disk` is how `_locate` is reached. + """ + src = Path(webrtc_server.__file__).read_text() + # Every call site, not the first one: a guard that stops at the first + # occurrence stops guarding the moment a new call is inserted above it. + calls = [m.start() for m in re.finditer(r"\bentry_abs_path\(", src)] + body = re.search(r"\ndef _locate\(.*?\n(?=\n\ndef |\n\nclass )", src, re.S) + assert body, "_locate is gone or has been renamed — this guard needs rewriting" + allowed = range(body.start(), body.end()) + stray = [c for c in calls if c not in allowed] + assert not stray, ( + f"{len(stray)} call(s) to entry_abs_path outside _locate: " + f"resolve a path through `off_disk(roots, _locate, ...)` instead") |