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
|
"""
Work on a group's roots belongs to the node, not to the session that asked for it.
Found live on 2026-09-14. A 900 GB directory was added from the desktop client,
so the op arrived over MNP and `_retarget_indexer` started the daemon's reload
with the session's own `_spawn`. That session closed 47 seconds later — the
client reconnected — and `shutdown_tasks()` cancelled everything it had started:
the reload, part-way through hashing the 42nd file, and the reload the
"removable" toggle had queued behind it. A cancelled task logs nothing. The new
root was in node.toml and in the indexer's own set, and never in the group's
context, so the node went on serving the old table for eight hours while
reconcile hashed the whole drive as "missed events" — and a client reloaded in
the morning showed nothing new. One loopback reload, which nothing could cancel,
put it right in nine milliseconds.
`test_root_ops_reach_the_live_set.py` covered this seam, with `_spawn` replaced
by a list: a fixture that could not cancel anything, testing a path whose whole
failure was being cancelled. These tests close the session for real.
"""
import asyncio
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node.config import load_config
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.roots import RootSet
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
pytestmark = pytest.mark.asyncio
GROUP = "g" * 32
def _dirs(tmp_path: Path) -> tuple[Path, Path]:
one, two = tmp_path / "one", tmp_path / "two"
one.mkdir()
two.mkdir()
(one / "a.txt").write_bytes(b"first root")
(two / "b.txt").write_bytes(b"second root, one")
(two / "c.txt").write_bytes(b"second root, two")
return one, two
def _set(*dirs: Path) -> RootSet:
return RootSet.build([{"path": str(d), "name": d.name} for d in dirs])
def _names(idx) -> list[str]:
return sorted(e.name for e in idx.index.entries)
async def _daemon(tmp_path: Path, one: Path, two: Path):
"""A daemon whose reload is held at the door, hosting one group whose
node.toml already names a second root the running set does not have."""
conf = tmp_path / "node.toml"
conf.write_text(
f'data_dir = "{(tmp_path / "data").as_posix()}"\n\n'
f'[[groups]]\nid = "{GROUP}"\nname = "plop"\n\n'
f' [[groups.roots]]\n path = "{one.as_posix()}"\n name = "one"\n\n'
f' [[groups.roots]]\n path = "{two.as_posix()}"\n name = "two"\n')
idx = DirectoryIndexer(roots=_set(one), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
await idx.initial_scan()
ctx = {"roots": idx.roots}
daemon = NodeDaemon.__new__(NodeDaemon)
daemon._config_path = conf
daemon._config = load_config(conf)
daemon._roster = None
daemon._hub = None
daemon._indexers = [idx]
daemon._reload_lock = asyncio.Lock()
daemon._state = {"groups_ctx": {GROUP: ctx}, "indexes": {}, "indexers": {},
"reload_fn": daemon._reload_config}
gate, entered = asyncio.Event(), asyncio.Event()
async def held_build_roots(group_cfg):
entered.set()
await gate.wait()
return await NodeDaemon._build_roots(daemon, group_cfg)
daemon._build_roots = held_build_roots
return daemon, idx, ctx, gate, entered
async def _until(predicate, timeout: float = 3.0) -> bool:
deadline = asyncio.get_running_loop().time() + timeout
while not predicate():
if asyncio.get_running_loop().time() > deadline:
return False
await asyncio.sleep(0.02)
return True
async def test_a_reload_started_over_mnp_survives_the_session_closing(tmp_path):
one, two = _dirs(tmp_path)
daemon, idx, ctx, gate, entered = await _daemon(tmp_path, one, two)
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
session._ctx = {"daemon_state": daemon._state}
session._tasks = set()
session._stop_stream = lambda: None
session._release_transfers = lambda: None
try:
await session._retarget_indexer(GROUP)
await asyncio.wait_for(entered.wait(), 2)
await session.shutdown_tasks() # what a closed client does
gate.set()
assert await _until(lambda: [r.name for r in ctx["roots"]] == ["one", "two"]), (
"closing the session that asked for the reload cancelled it")
assert await _until(lambda: not daemon._reload_lock.locked())
await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
assert _names(idx) == ["a.txt", "b.txt", "c.txt"]
finally:
gate.set()
await idx.stop()
async def test_a_reload_whose_caller_is_cancelled_still_finishes(tmp_path):
"""`ops.reload_config` awaits the reload inside the session's task too."""
one, two = _dirs(tmp_path)
daemon, idx, ctx, gate, entered = await _daemon(tmp_path, one, two)
try:
caller = asyncio.create_task(daemon._reload_config())
await asyncio.wait_for(entered.wait(), 2)
caller.cancel()
with pytest.raises(asyncio.CancelledError):
await caller
gate.set()
assert await _until(lambda: [r.name for r in ctx["roots"]] == ["one", "two"])
assert await _until(lambda: not daemon._reload_lock.locked())
finally:
gate.set()
await idx.stop()
async def test_a_plug_whose_caller_goes_away_does_not_empty_the_root(tmp_path):
one, two = _dirs(tmp_path)
class _Held(DirectoryIndexer):
gate = None
at_gate = None
async def _scan_root(self, root, **kwargs):
if self.gate is not None:
self.at_gate.set()
await self.gate.wait()
return await super()._scan_root(root, **kwargs)
idx = _Held(roots=_set(one, two), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
await idx.initial_scan()
idx.eject_root("two")
idx.gate, idx.at_gate = asyncio.Event(), asyncio.Event()
try:
caller = asyncio.create_task(idx.plug_root("two"))
await asyncio.wait_for(idx.at_gate.wait(), 2)
caller.cancel() # the admin op's session closed
with pytest.raises(asyncio.CancelledError):
await caller
idx.gate.set()
await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
assert _names(idx) == ["a.txt", "b.txt", "c.txt"], (
"the plug's rescan was cancelled after it had dropped the root's entries")
finally:
idx.gate.set()
await idx.stop()
|