1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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")
|