diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/conftest.py | 22 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_apps_enabled_policy.py | 9 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_cli_dispatch.py | 8 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_leaseless_reads.py | 85 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_media_cache_eviction.py | 152 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_partial_uploads.py | 489 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_platform.py | 10 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_stream_capacity.py | 155 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_transfer_settings.py | 152 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_transfer_slots.py | 365 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_transfer_slots_wire.py | 346 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 9 | ||||
| -rwxr-xr-x | packages/meshbay-node/tests/transfer_probe.py | 596 |
13 files changed, 2395 insertions, 3 deletions
diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py index ba86c13..692a118 100644 --- a/packages/meshbay-node/tests/conftest.py +++ b/packages/meshbay-node/tests/conftest.py @@ -17,6 +17,28 @@ needs_subprocess = pytest.mark.skipif( "SelectorEventLoop for aiortc", ) +@pytest.fixture(autouse=True) +def _restore_media_tool_paths(): + """Put `platform`'s resolved ffmpeg/ffprobe paths back after every test. + + `check_media_tools()` writes two module globals. `monkeypatch` restores what + a test patched, and knows nothing about what the code under test then wrote + — so a test that patched `shutil.which` to a Windows path and called + `check_media_tools()` left `_ffprobe_path` at "/opt/bin/ffprobe.exe" for the + rest of the session. Seven tests in two files about video transcoding then + died on FileNotFoundError, for a reason nowhere near themselves, and only + when the whole suite ran: run those two files alone and they passed. + + The instance is fixed at the call site as well; this closes the class. Any + future test that resolves media tools is undone here whether it remembers to + or not, which is the only way an order-dependent suite stops being one. + """ + from meshbay_node import platform as _plat + before = (_plat._ffmpeg_path, _plat._ffprobe_path) + yield + _plat._ffmpeg_path, _plat._ffprobe_path = before + + # Windows-only gaps still to close (see devel/windows-devel.md §5/§6). win32_todo = pytest.mark.skipif( sys.platform == "win32", diff --git a/packages/meshbay-node/tests/test_apps_enabled_policy.py b/packages/meshbay-node/tests/test_apps_enabled_policy.py index ac44ab3..40c7cc8 100644 --- a/packages/meshbay-node/tests/test_apps_enabled_policy.py +++ b/packages/meshbay-node/tests/test_apps_enabled_policy.py @@ -123,14 +123,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): "absent must mean every registered app, or an upgrade hides one " "for every existing group") await roster.set_enabled_apps("g1", ["chat"], set_by="op") - assert await roster.enabled_apps("g1") == ["chat"] + # Files comes back whatever was stored: `enabled_apps` inserts it at + # the front on read, and `ops.set_enabled_apps` does the same on write, + # because Settings is the one way back if everything else were turned + # off. The assertion predates that guard -- the code is right and the + # test was describing the older behaviour. + assert await roster.enabled_apps("g1") == ["files", "chat"] finally: await roster.close() reopened = Roster(db_path=tmp_path / "roster.db") await reopened.open() try: - assert await reopened.enabled_apps("g1") == ["chat"] + assert await reopened.enabled_apps("g1") == ["files", "chat"] assert sorted(await reopened.enabled_apps("g2")) == ["chat", "files"], ( "one group's setting must not answer for another") finally: diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py index 6f43772..64f96f1 100644 --- a/packages/meshbay-node/tests/test_cli_dispatch.py +++ b/packages/meshbay-node/tests/test_cli_dispatch.py @@ -46,6 +46,14 @@ VERBS = [ # that no longer takes one. ["member", "upload"], ["operator", "pair"], + ["transfers"], # defaults to show + ["transfers", "show"], + ["transfers", "set", "4", "2"], + ["transfers", "set", "4"], # only one number: usage, then exit + ["transfers", "set", "0", "2"], # zero is not "unlimited": refused + ["transfers", "per-member", "4", "2"], + ["transfers", "per-member", "4"], # only one number: usage, then exit + ["transfers", "per-member", "0", "2"], # zero is refused here too ["file", "list"], ["file", "rm", "abc", "--yes"], ["video", "rematch", "--yes"], diff --git a/packages/meshbay-node/tests/test_leaseless_reads.py b/packages/meshbay-node/tests/test_leaseless_reads.py new file mode 100644 index 0000000..70fd24f --- /dev/null +++ b/packages/meshbay-node/tests/test_leaseless_reads.py @@ -0,0 +1,85 @@ +""" +Browsing is never subject to a transfer slot — and is not unbounded either. + +**Operator decision, 2026-09-08:** a member must be able to browse a group that +is at capacity exactly as they browse an idle one. Not the poster grid, not the +covers, not opening a photo or a PDF to look at it. §3.4 of +~/next/improve-downloads.md satisfies that structurally: a transfer is what the +transfers widget shows, and nothing else takes a slot. + +But "not leased" cannot mean "unbounded". With MNP 3.0 making leases +compulsory, a client that simply omits `tr` would otherwise transfer outside +every cap, and the caps would be decoration — the leaseless branch left +reachable is finding C6's lesson (a transport that accepted a bare token) one +feature later. + +So a leaseless read is bounded by a small count of *files in flight*, not by +bytes: a RAW photo out of a camera is 60–80 MB and is browsing, a 40 MB archive +is a download, and no size threshold separates them. What separates them is +which function asked. +""" + +from meshbay_node.transfers import ( + LEASELESS_IDLE_SECS, MAX_LEASELESS_IN_FLIGHT, LeaselessReads, +) + + +def test_a_viewer_looking_at_one_file_is_never_refused(): + reads = LeaselessReads() + for chunk in range(20): + assert reads.admit("photo-1", now=float(chunk)) is True + + +def test_a_second_file_is_allowed_so_prefetching_stays_possible(): + """One is what a viewer needs; two is so the photo viewer can fetch the + next one while showing this one.""" + reads = LeaselessReads() + assert reads.admit("photo-1", now=0.0) is True + assert reads.admit("photo-2", now=0.0) is True + + +def test_a_third_file_is_refused(): + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + assert reads.admit("c", now=0.0) is False + + +def test_a_file_already_being_read_is_never_cut_off(): + """Even once the limit is reached. Refusing a chunk halfway through a photo + because the count moved would be worse than never having admitted it — the + viewer would show half an image and no error anyone can act on.""" + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + assert reads.admit("c", now=0.0) is False + assert reads.admit("a", now=1.0) is True + + +def test_finishing_one_frees_it_at_once(): + """The last chunk is the only "close" a leaseless read has. Waiting for the + idle timeout instead would mean somebody who looked at two photos cannot + look at a third for a minute.""" + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + reads.finish("a") + assert reads.admit("c", now=0.0) is True + + +def test_a_viewer_closed_mid_file_does_not_hold_its_place_for_ever(): + """It stops asking and says nothing — there is no message for "I closed the + tab". Without the idle expiry the session would carry two dead entries and + refuse every later preview, which is the bound turning into a bug.""" + reads = LeaselessReads() + reads.admit("a", now=0.0) + reads.admit("b", now=0.0) + assert reads.admit("c", now=1.0) is False + assert reads.admit("c", now=LEASELESS_IDLE_SECS + 2) is True + + +def test_the_bound_is_two(): + """Stated here so that changing it is a decision rather than a typo: it is + the number §3.4.1 argues for, and the argument is about viewers, not about + tuning.""" + assert MAX_LEASELESS_IN_FLIGHT == 2 diff --git a/packages/meshbay-node/tests/test_media_cache_eviction.py b/packages/meshbay-node/tests/test_media_cache_eviction.py new file mode 100644 index 0000000..587063a --- /dev/null +++ b/packages/meshbay-node/tests/test_media_cache_eviction.py @@ -0,0 +1,152 @@ +""" +The media cache has a ceiling, and reaching it drops the least useful rows. + +`thumbs` holds every generated thumbnail, every TMDB poster and backdrop, every +Cover Art Archive image and every cached audio transcode. Rows were removed only +when their source file left every group's index (`prune_file`), so a library +that merely *changes* over years grew this database with nothing to bound it. +Nothing in it is precious — every row is keyed off a value the node can +re-derive — which is what makes eviction the right answer rather than a bigger +disk. + +The migration is the part worth pinning hardest: `CREATE TABLE IF NOT EXISTS` +adds missing tables and never missing columns, so `used_at` would have reached a +fresh test database and never a deployed node — `CLAUDE.md`'s standing lesson +about `create_all()`. Every existing node has a `thumbs` table without it. +""" + +import sqlite3 + +import pytest + +from meshbay_node.media_cache import MediaCache + + +def _blob(n: int) -> bytes: + return b"x" * n + + +@pytest.mark.asyncio +async def test_the_cache_stays_under_its_cap(tmp_path): + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + cap = 40_000 + for i in range(20): + await cache.put_thumb(f"hash{i:03d}", f"file{i:03d}", _blob(5_000)) + await cache._evict_thumbs(cap=cap) + assert await cache.thumb_bytes() <= cap + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_what_is_evicted_is_what_nobody_asked_for(tmp_path): + """ + Least *recently used*, not least recently written: a poster fetched a year + ago and shown on every visit to a grid must outlive one cached last week and + never looked at again. + """ + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + for i in range(8): + await cache.put_thumb(f"hash{i}", f"file{i}", _blob(5_000)) + # The oldest row by write time, read now — so it is the newest by use. + assert await cache.get_thumb("hash0") is not None + await cache._evict_thumbs(cap=20_000) + assert await cache.get_thumb("hash0") is not None, ( + "evicted a row that had just been served") + assert await cache.get_thumb("hash1") is None, ( + "kept a row nothing had asked for since it was written") + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_a_lookup_by_synthetic_id_counts_as_use(tmp_path): + """ + `_fetch_and_cache_poster` finds an already-cached poster through + `get_thumb_hash_by_file_id`, which is the lookup a poster grid makes on + every visit. If that did not count as use, the images shown most often + would look like the coldest rows in the table. + """ + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + for i in range(8): + await cache.put_thumb(f"hash{i}", f"tmdb:/poster{i}.jpg", _blob(5_000)) + assert await cache.get_thumb_hash_by_file_id("tmdb:/poster0.jpg") == "hash0" + await cache._evict_thumbs(cap=20_000) + assert await cache.get_thumb("hash0") is not None + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_one_oversized_blob_does_not_empty_the_table(tmp_path): + """ + A single audio transcode larger than the whole cap would otherwise evict + everything and then itself, leaving an empty cache and the same problem. + """ + cache = MediaCache(db_path=tmp_path / "media_cache.db") + await cache.open() + try: + await cache.put_thumb("big", "file-big", _blob(50_000)) + removed = await cache._evict_thumbs(cap=10_000) + assert await cache.get_thumb("big") is not None + assert removed == 0 + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_an_existing_database_gains_the_column(tmp_path): + """ + The migration, against a database shaped exactly like a deployed node's: + `thumbs` with no `used_at`, holding a row that must survive. + """ + db_path = tmp_path / "media_cache.db" + con = sqlite3.connect(db_path) + con.executescript(""" + CREATE TABLE thumbs ( + thumb_hash TEXT PRIMARY KEY, + file_id TEXT NOT NULL, + jpeg BLOB NOT NULL + ); + CREATE INDEX idx_thumbs_file ON thumbs(file_id); + """) + con.execute("INSERT INTO thumbs VALUES (?, ?, ?)", ("old", "file-old", b"abc")) + con.commit() + con.close() + + cache = MediaCache(db_path=db_path) + await cache.open() + try: + assert await cache.get_thumb("old") == b"abc", "the migration lost a row" + # Seeded with "now", not 0: an upgrade must not make every existing row + # look infinitely old and evict the whole cache on the next write. + con = sqlite3.connect(db_path) + used_at = con.execute( + "SELECT used_at FROM thumbs WHERE thumb_hash = 'old'").fetchone()[0] + con.close() + assert used_at > 0, "existing rows were left at 0 and are first to go" + finally: + await cache.close() + + +@pytest.mark.asyncio +async def test_opening_twice_is_harmless(tmp_path): + """The migration must be idempotent — a node opens this on every start.""" + db_path = tmp_path / "media_cache.db" + for _ in range(3): + cache = MediaCache(db_path=db_path) + await cache.open() + await cache.put_thumb("h", "f", b"xyz") + await cache.close() + cache = MediaCache(db_path=db_path) + await cache.open() + try: + assert await cache.get_thumb("h") == b"xyz" + finally: + await cache.close() diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py new file mode 100644 index 0000000..f5b6602 --- /dev/null +++ b/packages/meshbay-node/tests/test_partial_uploads.py @@ -0,0 +1,489 @@ +""" +Two rules about an upload that stopped in the middle. + +**It belongs to the group, not to the connection.** Progress used to be kept on +the session, so a dropped connection lost it and the client's next chunk was +refused with `not_started` — an upload interrupted at 99% could only start again +from zero, on a link flaky enough to have interrupted it once. + +**And what it leaves on disk has an owner or it has an end.** The state that was +lost left a `.part` file nothing would ever finish, delete or look at again: +invisible in the index, because `.part` is not an index entry, and a gigabyte of +somebody else's disk for one abandoned film. + +The keying is a correctness property rather than a nicety: a shared directory +means two members can be sending `IMG_1234.jpg` at the same moment, and neither +may inherit — or overwrite the position of — the other's. +""" + +import os +import time +import types +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import Root, RootSet +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from meshbay_common.protocol import ( + UPLOAD_PROBE_INDEX, file_upload_ack_payload, +) + +from conftest import one_root, sealed_upload + +from meshbay_node.uploads import ( + ORPHAN_AFTER_SECS, PART_SUFFIX, PartialUploads, find_parts, orphaned_parts, +) + + +# ── the state ─────────────────────────────────────────────────────────────── + +def test_an_upload_is_found_again_after_the_connection_went_away(): + """The whole point: the store outlives the session, so the position is + still there when the client comes back.""" + uploads = PartialUploads() + uploads.start("alice", "media", "film.mkv", "film.mkv") + uploads.advance("alice", "media", "film.mkv", chunk_index=0, nbytes=1024) + uploads.advance("alice", "media", "film.mkv", chunk_index=1, nbytes=1024) + + state = uploads.get("alice", "media", "film.mkv") + assert state is not None + assert state.next_index == 2 + assert state.bytes == 2048 + + +def test_two_members_uploading_the_same_name_do_not_share_a_position(): + """A shared folder makes this ordinary, not adversarial: everyone's camera + produces the same filenames. Inheriting the other's position would append + one person's chunks to another person's file.""" + uploads = PartialUploads() + uploads.start("alice", "photos", "IMG_1234.jpg", "IMG_1234.jpg") + uploads.start("bob", "photos", "IMG_1234.jpg", "IMG_1234 (2).jpg") + uploads.advance("alice", "photos", "IMG_1234.jpg", 0, 10) + + assert uploads.get("alice", "photos", "IMG_1234.jpg").next_index == 1 + assert uploads.get("bob", "photos", "IMG_1234.jpg").next_index == 0 + assert uploads.get("bob", "photos", "IMG_1234.jpg").stored_name \ + == "IMG_1234 (2).jpg" + + +def test_the_same_name_in_two_directories_is_two_uploads(): + uploads = PartialUploads() + uploads.start("alice", "media", "a.bin", "a.bin") + uploads.start("alice", "archive", "a.bin", "a.bin") + uploads.advance("alice", "media", "a.bin", 0, 5) + assert uploads.get("alice", "archive", "a.bin").next_index == 0 + + +def test_advancing_an_upload_nobody_started_says_so(): + """The caller refuses the chunk on this; silently creating the state here + would let a client append to whatever `.part` is already on disk.""" + assert PartialUploads().advance("alice", "media", "x", 0, 1) is None + + +def test_starting_again_forgets_the_old_position(): + """Chunk zero means "from the beginning" — the file is opened for writing, + not appending, so the position has to go with it.""" + uploads = PartialUploads() + uploads.start("alice", "media", "a.bin", "a.bin") + uploads.advance("alice", "media", "a.bin", 0, 500) + uploads.start("alice", "media", "a.bin", "a.bin") + assert uploads.get("alice", "media", "a.bin").next_index == 0 + assert uploads.get("alice", "media", "a.bin").bytes == 0 + + +# ── the reaper ────────────────────────────────────────────────────────────── + +def _old(seconds: float) -> float: + return 1_000_000.0 - seconds + + +NOW = 1_000_000.0 +FILM = Path("/roots/media/film.mkv.part") + + +def test_a_part_nobody_is_writing_and_nobody_has_touched_is_deleted(): + """The leak this exists to close: an abandoned upload's file, kept for ever + and invisible because `.part` is not an index entry.""" + doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS + 1))], + live=set(), now=NOW) + assert doomed == [FILM] + + +def test_an_upload_in_progress_is_never_deleted(): + """Even when its file is old: a large upload over a slow link is exactly the + one that has been on disk the longest, and it is the one that would hurt + most to lose.""" + doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS * 3))], + live={FILM}, now=NOW) + assert doomed == [] + + +def test_a_recently_written_part_is_left_alone(): + """No state and recent writes is a client that has just reconnected, or one + whose state this node has not seen yet. Waiting a day costs disk; being + wrong costs somebody their upload.""" + doomed = orphaned_parts([(FILM, _old(60))], live=set(), now=NOW) + assert doomed == [] + + +def test_the_same_name_in_another_directory_does_not_protect_it(): + """Matched on the whole path, so an upload to `media/` cannot keep an + orphan in `archive/` alive for ever. Comparing names would; comparing a + path rebuilt from a root and a relative directory would be a second + implementation that has to agree with the first for ever, and the state + records the path it is writing instead.""" + other = Path("/roots/archive/film.mkv.part") + doomed = orphaned_parts([(other, _old(ORPHAN_AFTER_SECS + 1))], + live={FILM}, now=NOW) + assert doomed == [other] + + +def test_a_finished_file_is_not_a_candidate(): + """Only `.part` is ever deleted. A bug that let this touch a real file would + be the worst one in the project, so the check is here as well as at the call + site that only offers `.part` paths.""" + doomed = orphaned_parts( + [(Path("/roots/media/film.mkv"), _old(ORPHAN_AFTER_SECS * 10))], + live=set(), now=NOW) + assert doomed == [] + + +def test_a_file_from_the_future_is_left_alone(): + """A clock that went backwards is not evidence that a file is abandoned, and + deleting is not reversible.""" + doomed = orphaned_parts([(Path("/roots/media/a.part"), NOW + 10_000)], + live=set(), now=NOW) + assert doomed == [] + + +def test_the_boundary_is_the_age_itself(): + at = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS))] + just_under = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS - 1))] + assert orphaned_parts(at, set(), NOW) == [Path("/roots/media/a.part")] + assert orphaned_parts(just_under, set(), NOW) == [] + + +def test_an_upload_records_the_file_it_is_writing(): + """What keeps the reaper honest. Without it the two sides would have to + agree on how a path is built from a root name and a relative directory — + two implementations of one rule, and the failure mode is deleting a live + upload.""" + uploads = PartialUploads() + uploads.start("alice", "media", "film.mkv", "film.mkv", part_path=FILM) + assert uploads.live_paths() == {FILM} + uploads.drop("alice", "media", "film.mkv") + assert uploads.live_paths() == set() + + +def test_the_suffix_is_named_once(): + """Two spellings of `.part` would be a bug nobody could see: the writer + would produce one and the reaper would look for the other.""" + assert PART_SUFFIX == ".part" + + +# ── the walk, and the deletion ────────────────────────────────────────────── + +def _root(tmp_path, name, *, writable=True, available=True) -> Root: + path = tmp_path / name + path.mkdir(parents=True, exist_ok=True) + return Root(name=name, path=path, writable=writable, available=available) + + +def _aged(path: Path, seconds: float, content: bytes = b"x") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + when = time.time() - seconds + os.utime(path, (when, when)) + return path + + +def test_the_walk_finds_parts_in_subdirectories(tmp_path): + """Uploads go into the folder the sender was looking at, which is any + directory in the group — not a quarantine subfolder, since 2026-08-14.""" + root = _root(tmp_path, "media") + _aged(root.path / "a.part", 10) + _aged(root.path / "series" / "b.part", 10) + _aged(root.path / "series" / "kept.mkv", 10) + found = {p.name for p, _ in find_parts([root])} + assert found == {"a.part", "b.part"} + + +def test_a_read_only_root_is_not_walked(tmp_path): + """It cannot have received an upload, so anything `.part` in it belongs to + the operator and is none of this code's business.""" + root = _root(tmp_path, "library", writable=False) + _aged(root.path / "theirs.part", ORPHAN_AFTER_SECS * 2) + assert find_parts([root]) == [] + + +def test_an_unavailable_root_is_not_walked(tmp_path): + """A drive that is not mounted. Walking it finds nothing, and "nothing + found" is the input from which a careless janitor concludes everything is + gone.""" + root = _root(tmp_path, "external", available=False) + _aged(root.path / "x.part", ORPHAN_AFTER_SECS * 2) + assert find_parts([root]) == [] + + +def _daemon(groups: dict) -> NodeDaemon: + """A daemon with nothing but what `_reap_once` reads.""" + daemon = NodeDaemon.__new__(NodeDaemon) + daemon._webrtc = types.SimpleNamespace(_ctx={"groups": groups}) + return daemon + + +def test_the_janitor_deletes_the_abandoned_and_keeps_the_rest(tmp_path): + """End to end on real files: the old orphan goes, the recent one and the + one somebody is still writing stay, and a finished file is never a + candidate.""" + root = _root(tmp_path, "media") + old = _aged(root.path / "abandoned.mkv.part", ORPHAN_AFTER_SECS + 60) + recent = _aged(root.path / "fresh.mkv.part", 30) + live = _aged(root.path / "sending.mkv.part", ORPHAN_AFTER_SECS * 2) + finished = _aged(root.path / "done.mkv", ORPHAN_AFTER_SECS * 5) + + uploads = PartialUploads() + uploads.start("alice", "media", "sending.mkv", "sending.mkv", part_path=live) + + daemon = _daemon({"g1": {"roots": RootSet(roots=[root]), + "partial_uploads": uploads}}) + assert daemon._reap_once() == 1 + assert not old.exists() + assert recent.exists() and live.exists() and finished.exists() + + +def test_a_group_that_has_never_uploaded_anything_is_handled(tmp_path): + """No `partial_uploads` in the context yet — it is created on first use, so + a node that has been up for five minutes has none.""" + root = _root(tmp_path, "media") + old = _aged(root.path / "left.mkv.part", ORPHAN_AFTER_SECS + 1) + daemon = _daemon({"g1": {"roots": RootSet(roots=[root])}}) + assert daemon._reap_once() == 1 + assert not old.exists() + + +def test_a_group_with_no_roots_is_skipped(tmp_path): + assert _daemon({"g1": {}})._reap_once() == 0 + + +# ── across two connections ────────────────────────────────────────────────── + +GROUP = "g" * 32 + + +def _peer(ctx: dict, user_id: str = "user-1") -> WebRTCPeerSession: + """One connection into a group whose context is shared, as it is on a node. + + Two of these standing for the same member is the whole point: the second is + the reconnection, and it must find what the first was doing. + """ + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = GROUP + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _group_ctx(tmp_path) -> dict: + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + return {"roots": one_root(shared), + "index": GroupIndex(group_id=GROUP, + sk_node=Ed25519PrivateKey.generate()), + "gek": generate_gek()} + + +def _errors(session): + return [m for m in session.sent if m.get("type") == "error"] + + +def test_an_upload_survives_the_connection_that_started_it(tmp_path): + """The defect this stage exists to fix. + + The state used to live on the session, so the second connection saw no + upload at all and refused the chunk with `not_started`: an upload + interrupted at 99% could only be started again from zero, on a link flaky + enough to have interrupted it once. + """ + ctx = _group_ctx(tmp_path) + first = _peer(ctx) + first._do_file_upload(sealed_upload(first, filename="film.mkv", + data=b"first-half", + chunk_index=0, total_chunks=2)) + assert _errors(first) == [] + + # The link drops; the client comes back on a new connection and carries on. + second = _peer(ctx) + second._do_file_upload(sealed_upload(second, filename="film.mkv", + data=b"second-half", + chunk_index=1, total_chunks=2)) + assert _errors(second) == [], _errors(second) + + root = ctx["roots"].roots[0] + assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half" + + +def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): + """The key includes the member for a reason. Without it, a second person + sending the same name into the same folder would append their chunks to the + first person's file — which a shared folder makes an ordinary accident, not + only an attack.""" + ctx = _group_ctx(tmp_path) + alice = _peer(ctx, "alice") + alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", + data=b"hers", chunk_index=0, + total_chunks=2)) + assert _errors(alice) == [] + + bob = _peer(ctx, "bob") + bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", + data=b"his", chunk_index=1, + total_chunks=2)) + assert [m.get("code") for m in _errors(bob)] == ["not_started"] + + +def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): + """The two halves of this stage meeting: the state the node keeps is what + stops the janitor deleting a file somebody is still sending.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="film.mkv", + data=b"half", chunk_index=0, + total_chunks=2)) + live = ctx["partial_uploads"].live_paths() + assert len(live) == 1 + assert next(iter(live)).name == "film.mkv.part" + assert next(iter(live)).exists() + + +# ── asking where to resume ────────────────────────────────────────────────── + + +def _acks(session, ctx): + return [file_upload_ack_payload(ctx["gek"], GROUP, m) + for m in session.sent if m.get("type") == "file_upload_ack"] + + +def _probe(session, filename: str) -> dict: + """The question, asked exactly as the client asks it: an ordinary sealed + upload chunk with no bytes and the probe index.""" + return sealed_upload(session, filename=filename, data=b"", + chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1) + + +def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(_probe(peer, "film.mkv")) + assert _errors(peer) == [] + assert _acks(peer, ctx)[0]["resume_from"] == 0 + + +def test_a_probe_reports_what_the_node_already_holds(tmp_path): + """The point of the whole stage: the client learns it has 2 chunks there and + sends the third, instead of sending a film again.""" + ctx = _group_ctx(tmp_path) + first = _peer(ctx) + for i in range(2): + first._do_file_upload(sealed_upload(first, filename="film.mkv", + data=b"xxxx", chunk_index=i, + total_chunks=5)) + assert _errors(first) == [] + + reconnected = _peer(ctx) + reconnected._do_file_upload(_probe(reconnected, "film.mkv")) + ack = _acks(reconnected, ctx)[0] + assert ack["resume_from"] == 2 + assert ack["stored_as"] == "film.mkv" + + +def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): + """It has to be free of consequence: a client that asks and goes away must + leave no file, no state and no name taken.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(_probe(peer, "film.mkv")) + root = ctx["roots"].roots[0] + assert list(root.path.iterdir()) == [] + assert len(ctx.get("partial_uploads") or []) == 0 + # And it promises no destination it has not taken. + assert _acks(peer, ctx)[0]["stored_as"] == "" + + +def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): + """Same keying as the upload itself. Otherwise one member could measure + another's progress on a file they never sent — and worse, resume it.""" + ctx = _group_ctx(tmp_path) + alice = _peer(ctx, "alice") + alice._do_file_upload(sealed_upload(alice, filename="film.mkv", + data=b"xxxx", chunk_index=0, + total_chunks=5)) + bob = _peer(ctx, "bob") + bob._do_file_upload(_probe(bob, "film.mkv")) + assert _acks(bob, ctx)[0]["resume_from"] == 0 + + +def test_an_ordinary_ack_carries_no_resume_field(tmp_path): + """So a client can tell a probe's answer from a chunk's without looking at + the index it echoed.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", + chunk_index=0, total_chunks=2)) + assert "resume_from" not in _acks(peer, ctx)[0] + + +def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): + """Every check the write path makes has already run when the probe is + answered, so it cannot be used to ask questions about somewhere the caller + may not write.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="../escape", + data=b"", chunk_index=UPLOAD_PROBE_INDEX, + total_chunks=1)) + assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"] + assert _acks(peer, ctx) == [] + + +# ── the slot an upload holds ──────────────────────────────────────────────── + +def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): + """A grant nobody takes up is reclaimed after thirty seconds and abandoned + on the third miss. Uploads are not gated by the lease, so the file arrived + anyway — but the widget follows the lease, and a 3.5 GB upload therefore + read "waiting, 0 ahead" for a minute and a half while it was transferring, + with three reclaims logged against it. + + The download twin of this was fixed a day earlier; the same omission was + still here, invisible until uploads took a real lease. + """ + from meshbay_node.transfers import TransferSlots, UPLOAD + + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + slots = TransferSlots() + peer._ctx = dict(ctx) + peer._ctx["_transfer_slots"] = slots + peer._registry_key = "session-1" + lease, err = slots.open(tr="up-1", kind=UPLOAD, session_key="session-1", + user_id="user-1", group_id=GROUP, bytes=10, chunks=2) + assert not err and lease.state == "granted" + assert lease.used is False + + msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx", + chunk_index=0, total_chunks=2) + msg["tr"] = "up-1" + peer._do_file_upload(msg) + + assert _errors(peer) == [] + assert slots.leases["up-1"].used is True, ( + "the node still believes nobody took this slot up, and will reclaim it") diff --git a/packages/meshbay-node/tests/test_platform.py b/packages/meshbay-node/tests/test_platform.py index 92e74df..3fb27f3 100644 --- a/packages/meshbay-node/tests/test_platform.py +++ b/packages/meshbay-node/tests/test_platform.py @@ -71,6 +71,10 @@ def test_check_media_tools_raises_when_ffmpeg_is_missing(monkeypatch): def test_check_media_tools_stores_the_resolved_paths(monkeypatch): + # This call writes two module globals, and what undoes them is the autouse + # `_restore_media_tool_paths` fixture in conftest.py -- see it for what went + # wrong when nothing did. Deliberately not repeated here: one mechanism, one + # explanation, or the two drift. monkeypatch.setattr(plat.shutil, "which", lambda n: f"/opt/bin/{n}.exe") plat.check_media_tools("ffmpeg", "ffprobe") @@ -365,6 +369,9 @@ def test_service_install_uses_s4u_not_a_stored_password(monkeypatch): credential validation, and omitting /rp registers "Interactive only", which never runs at boot or on demand. See platform.py's service mode comment for the full story.""" + # Service mode is Windows-only and refuses outright anywhere else; + # every other test in this file says so, these two never did. + monkeypatch.setattr(sys, "platform", "win32") monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user") calls = [] monkeypatch.setattr( @@ -388,6 +395,9 @@ def test_service_install_tolerates_no_startup_launcher_present(win_startup, monk def test_service_install_raises_with_powershells_error_message(monkeypatch): + # Service mode is Windows-only and refuses outright anywhere else; + # every other test in this file says so, these two never did. + monkeypatch.setattr(sys, "platform", "win32") monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user") monkeypatch.setattr( plat.subprocess, "run", diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py new file mode 100644 index 0000000..a35ece8 --- /dev/null +++ b/packages/meshbay-node/tests/test_stream_capacity.py @@ -0,0 +1,155 @@ +""" +`max_concurrent_streams` must take effect without a restart. + +`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an +attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so +`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the +setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the +Node page offers it as a live setting, and it did nothing: an operator lowering +the cap on a struggling machine, or raising it after "Server busy", saw no +change and had no way to know why. + +Nothing here mocks the pool. `set_capacity` is called on a real +`WebRTCTransport` and the assertions read what a stream request would actually +find. +""" + +import asyncio + +import pytest + +from meshbay_node.transport.webrtc_server import ( + MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, WebRTCTransport, +) + + +def _pool(transport) -> asyncio.Semaphore: + """The pool a stream request would acquire, built the way one builds it.""" + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = transport._ctx + return session._transcode_semaphore() + + +@pytest.fixture +def transport(tmp_path): + """A real WebRTCTransport. Its keys and index are genuine but incidental — + nothing below the capacity code reads them.""" + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + from conftest import one_root + from meshbay_common.crypto import generate_gek + from meshbay_node.indexer.group_index import GroupIndex + + sk_node = Ed25519PrivateKey.generate() + gek = generate_gek() + shared = tmp_path / "shared" + shared.mkdir() + return WebRTCTransport( + sk_node=sk_node, hub_pk_pem=b"", gek=gek, + roots=one_root(shared), + index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek), + stun_servers=[]) + + +def test_raising_the_cap_is_visible_to_the_next_stream(transport): + """The bug, at its simplest: the number changes and nothing happens.""" + pool = _pool(transport) + assert pool._value == MAX_CONCURRENT_TRANSCODES + transport.set_capacity(max_concurrent_streams=16) + assert _pool(transport)._value == 16, ( + "the setting was accepted and the pool never changed — this is the " + "no-op that shipped") + + +def test_lowering_the_cap_does_not_interrupt_what_is_running(transport): + """ + A slot is held for the length of a film, so lowering the cap cannot take a + viewer's film away. It stops the next one starting, and the replacement pool + carries only the permits that remain. + """ + _pool(transport) + transport._ctx["_streams_in_flight"] = 3 + transport.set_capacity(max_concurrent_streams=4) + assert _pool(transport)._value == 1, ( + "a full set of permits would let more viewers in than either the old " + "cap or the new one, on top of the three still watching") + + +def test_lowering_below_what_is_running_refuses_the_next_one(transport): + _pool(transport) + transport._ctx["_streams_in_flight"] = 6 + transport.set_capacity(max_concurrent_streams=2) + assert _pool(transport)._value == 0, "the pool must not go negative" + + +def test_the_value_is_kept_for_a_pool_not_yet_built(transport): + """Nothing has streamed, so there is nothing to resize — but the number has + to be there when the first request builds the pool.""" + transport.set_capacity(max_concurrent_streams=3) + assert transport._ctx.get("_transcode_sem") is None + assert _pool(transport)._value == 3 + + +def test_a_cap_below_one_is_refused(transport): + for bad in (0, -1): + with pytest.raises(ValueError): + transport.set_capacity(max_concurrent_streams=bad) + + +def test_nothing_changes_when_nothing_is_passed(transport): + _pool(transport) + before = transport._ctx["_transcode_sem"] + assert transport.set_capacity() == {} + assert transport._ctx["_transcode_sem"] is before + + +@pytest.mark.asyncio +async def test_in_flight_is_counted_by_the_streaming_path_itself(transport): + """ + `set_capacity` resizes against `_streams_in_flight`, so that counter has to + be maintained where slots are actually taken — not set by a test. Drives the + real `_stream_video`, with the work under it stubbed: what is being checked + is the accounting around the slot, which is where flow control in this repo + has gone wrong before. + """ + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = transport._ctx + session._send = lambda msg: None + + seen = [] + release = asyncio.Event() + + async def _inner(_msg): + seen.append(transport._ctx.get("_streams_in_flight")) + await release.wait() + + session._stream_video_inner = _inner + task = asyncio.create_task(session._stream_video({"file_id": "x"})) + await asyncio.sleep(0) + await asyncio.sleep(0) + assert seen == [1], "the slot was taken without being counted" + + release.set() + await task + assert transport._ctx["_streams_in_flight"] == 0, ( + "a slot that is not given back is a viewer nobody can replace — the " + "class of bug _replace_stream and shutdown_tasks exist for") + + +def test_ops_calls_the_real_mechanism(): + """ + The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every + WebRTCTransport that has ever existed, so a test that only checked + "set_node_settings does not raise" passed throughout. + """ + import inspect + + from meshbay_node import ops + + src = inspect.getsource(ops.set_node_settings) + # Comments stripped: this function now *explains* the dead attribute, and a + # test that matched the prose would fail on its own documentation. + code = "\n".join(line.split("#", 1)[0] for line in src.splitlines()) + assert "_stream_sem" not in code, "the attribute that never existed is back" + assert "set_capacity" in code, "the setting must reach the pool that exists" + assert not hasattr(WebRTCTransport, "_stream_sem") diff --git a/packages/meshbay-node/tests/test_transfer_settings.py b/packages/meshbay-node/tests/test_transfer_settings.py new file mode 100644 index 0000000..79502e7 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_settings.py @@ -0,0 +1,152 @@ +""" +The two scopes a transfer cap has, and the rule that they are not the same kind +of setting. + +The **pools** are the machine's: how many transfers this node runs at once, +across every group, from `[node]` in node.toml with a roster override — the +§2.11 pattern, changed from the Node page or the CLI, applied live. + +The **per-member cap** is a group's: how many one member may run at once here. +It lives on the node like every other group setting (not the hub, which would +have authority over someone else's disk; not node.toml, which is hand-written +and needs a restart), and changing it is a signed operator instruction, because +an unsigned cap is one any member can raise for themselves. + +What is checked here is the seam between the stored value and the pool that +enforces it — a setting that is written, acknowledged and never read is the +shape of the bug this whole branch started from (`webrtc._stream_sem`). +""" + +import pytest + +from meshbay_node.roster import Roster +from meshbay_node.transfers import ( + DEFAULT_MAX_PER_MEMBER, DOWNLOAD, UPLOAD, TransferSlots, +) + + +@pytest.fixture +async def roster(tmp_path): + r = Roster(db_path=tmp_path / "roster.db") + await r.open() + yield r + await r.close() + + +# ── the group's own cap ───────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_absent_means_the_default_not_unlimited(roster): + """A group that predates the setting must not come back unlimited: the + node-wide pool would then be the only control, which is the situation slots + exist to end.""" + assert await roster.transfer_limits("g1") == {} + slots = TransferSlots() + assert slots.member_cap(DOWNLOAD, ("g1", "alice")) == DEFAULT_MAX_PER_MEMBER + + +@pytest.mark.asyncio +async def test_the_cap_survives_a_restart(roster, tmp_path): + await roster.set_transfer_limits("g1", {"download": 4, "upload": 1}, + set_by="op") + await roster.close() + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.transfer_limits("g1") == {"download": 4, + "upload": 1} + finally: + await reopened.close() + + +@pytest.mark.asyncio +async def test_one_group_does_not_set_anothers(roster): + await roster.set_transfer_limits("g1", {"download": 5}, set_by="op") + assert await roster.transfer_limits("g2") == {} + + +@pytest.mark.asyncio +async def test_a_stored_zero_never_becomes_a_cap_of_zero(roster): + """Zero is not "unlimited" and must not be "nobody may transfer" either. + Whatever reaches storage, the floor is one.""" + await roster.set_transfer_limits("g1", {"download": 0}, set_by="op") + assert (await roster.transfer_limits("g1"))["download"] == 1 + + +@pytest.mark.asyncio +async def test_rubbish_in_the_row_reads_as_unset(roster): + """A row this code did not write must not take a group's transfers down — + the same "a payload that does not open ends nothing silently" discipline + the sealed messages follow.""" + await roster.set_setting("g1", Roster.SETTING_TRANSFER_LIMITS, + "not json", "op") + assert await roster.transfer_limits("g1") == {} + + +# ── the pool that enforces it ─────────────────────────────────────────────── + +def test_a_group_cap_overrides_the_node_default(): + slots = TransferSlots() + slots.set_group_limits("strict", {DOWNLOAD: 1}) + assert slots.member_cap(DOWNLOAD, ("strict", "alice")) == 1 + assert slots.member_cap(DOWNLOAD, ("other", "alice")) == DEFAULT_MAX_PER_MEMBER + assert slots.member_cap(UPLOAD, ("strict", "alice")) == DEFAULT_MAX_PER_MEMBER, ( + "setting the download cap must not silently change the upload one") + + +def test_the_group_cap_is_what_queues_a_member(): + slots = TransferSlots() + slots.set_group_limits("strict", {DOWNLOAD: 1}) + args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", + group_id="strict") + assert slots.open(tr="t1", **args)[0].state == "granted" + assert slots.open(tr="t2", **args)[0].state == "queued" + + +def test_raising_a_group_cap_starts_what_was_waiting(): + slots = TransferSlots() + slots.set_group_limits("g1", {DOWNLOAD: 1}) + args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", group_id="g1") + slots.open(tr="t1", **args) + slots.open(tr="t2", **args) + granted = slots.set_group_limits("g1", {DOWNLOAD: 3}) + assert [x.tr for x in granted] == ["t2"], ( + "the cap was raised and the waiting transfer was left waiting") + + +def test_one_groups_cap_does_not_move_anothers_queue(): + slots = TransferSlots() + slots.set_group_limits("g1", {DOWNLOAD: 1}) + slots.set_group_limits("g2", {DOWNLOAD: 1}) + for g in ("g1", "g2"): + args = dict(kind=DOWNLOAD, session_key=f"s-{g}", user_id="alice", + group_id=g) + slots.open(tr=f"{g}-1", **args) + slots.open(tr=f"{g}-2", **args) + granted = slots.set_group_limits("g1", {DOWNLOAD: 2}) + assert [x.tr for x in granted] == ["g1-2"] + assert slots.leases["g2-2"].state == "queued" + + +# ── the node-wide pools ───────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_the_node_wide_caps_round_trip_through_the_roster(roster): + defaults = {"max_concurrent_downloads": 8, "max_concurrent_uploads": 8} + assert await roster.node_settings(defaults) == { + **{k: v for k, v in defaults.items()}, + **{k: None for k in ("invite_ttl_hours", "pair_ttl_hours", + "device_request_ttl_minutes", + "max_concurrent_streams", + "transcode_incompatible_video")}, + "stun_servers": [], "ice_interfaces": [], + } + await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op") + assert (await roster.node_settings(defaults))["max_concurrent_downloads"] == 3 + + +def test_node_toml_carries_both_keys(): + """The template is what an operator reads before they read any document.""" + from meshbay_node.config import EXAMPLE_CONFIG as tpl + assert "max_concurrent_downloads" in tpl + assert "max_concurrent_uploads" in tpl diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py new file mode 100644 index 0000000..7056e93 --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -0,0 +1,365 @@ +""" +Transfer slots: the caps, the queue, and every way a slot can be lost. + +The requirement this is written against is not "a cap exists". It is that +**nobody stays stuck** — neither a slot the node never gets back, which fills +the node and queues everyone for ever, nor a transfer a client shows as waiting +that the node has already forgotten. + +`TransferSlots` has no asyncio and no transport in it precisely so that those +failures can be driven here instead of through a DataChannel, where they are +rare, timing-dependent and unprovable. The clock is passed in, so the two +timeouts are exercised without a test that sleeps for two minutes. + +`test_the_counter_never_drifts` is the one that matters most: a stuck slot is a +race by nature, "it works now" is not evidence against a race, and the earlier +flow-control bugs in this repo (window_leak.mjs, the discarded segment that +leaked a slot per discard) were all found by forcing the worst case rather than +by reasoning about it. +""" + +import random + +import pytest + +from meshbay_node.transfers import ( + DOWNLOAD, GRANT_DEADLINE_SECS, IDLE_TIMEOUT_SECS, KINDS, + MAX_MISSED_GRANTS, MAX_QUEUED_PER_MEMBER, REASON_ABANDONED, REASON_IDLE, + REASON_NOT_TAKEN_UP, TransferSlots, + UPLOAD, +) + + +def _slots(node=8, per_member=2) -> TransferSlots: + s = TransferSlots() + s.caps = {k: node for k in KINDS} + s.per_member = {k: per_member for k in KINDS} + return s + + +def _open(s, tr, *, session="s1", user="u1", group="g1", kind=DOWNLOAD, now=0.0): + lease, err = s.open(tr=tr, kind=kind, session_key=session, user_id=user, + group_id=group, now=now) + assert not err, err + return lease + + +# ── the caps ──────────────────────────────────────────────────────────────── + +def test_a_member_is_held_to_their_own_cap_first(_=None): + s = _slots(node=8, per_member=2) + assert _open(s, "a").state == "granted" + assert _open(s, "b").state == "granted" + assert _open(s, "c").state == "queued", ( + "a third transfer for one member must queue even though the node has " + "six free slots — otherwise one member takes the node") + + +def test_the_member_cap_spans_their_devices(_=None): + """Per account, not per connection: two browsers and a desktop client + signed in as the same person share the two slots, or the cap becomes a + function of how many tabs somebody opens.""" + s = _slots(per_member=2) + _open(s, "a", session="laptop") + _open(s, "b", session="phone") + assert _open(s, "c", session="desktop").state == "queued" + + +def test_the_node_cap_holds_across_members(_=None): + s = _slots(node=3, per_member=2) + _open(s, "a", user="u1") + _open(s, "b", user="u1") + _open(s, "c", user="u2") + assert _open(s, "d", user="u2").state == "queued" + assert s.in_use(DOWNLOAD) == 3 + + +def test_downloads_and_uploads_have_separate_pools(_=None): + s = _slots(node=2, per_member=2) + _open(s, "a", kind=DOWNLOAD) + _open(s, "b", kind=DOWNLOAD) + assert _open(s, "c", kind=UPLOAD).state == "granted", ( + "a full download pool must not stop an upload") + + +# ── the queue ─────────────────────────────────────────────────────────────── + +def test_a_freed_slot_goes_to_whoever_was_waiting(_=None): + s = _slots(node=1, per_member=2) + _open(s, "a", user="u1") + queued = _open(s, "b", user="u2") + assert queued.state == "queued" + _, granted = s.close("a") + assert [x.tr for x in granted] == ["b"] + assert s.leases["b"].state == "granted" + + +def test_a_member_at_their_cap_is_skipped_not_waited_for(_=None): + """Granting strictly in arrival order lets one member's own limit stall + every other member behind them.""" + s = _slots(node=3, per_member=2) + _open(s, "a", user="u1") + _open(s, "b", user="u1") + hog = _open(s, "c", user="u1") # u1 is at their cap + other = _open(s, "d", user="u2") # arrives later + assert hog.state == "queued" + assert other.state == "granted", "u2 was made to wait behind u1's own limit" + + +def test_position_is_reported_from_the_queue_itself(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a") + b, c = _open(s, "b"), _open(s, "c") + assert (s.ahead_of(b), s.ahead_of(c)) == (0, 1) + + +def test_a_member_cannot_queue_without_end(_=None): + s = _slots(node=1, per_member=1) + _open(s, "granted") + for i in range(MAX_QUEUED_PER_MEMBER): + _open(s, f"q{i}") + lease, err = s.open(tr="one-too-many", kind=DOWNLOAD, session_key="s1", + user_id="u1", group_id="g1") + assert lease is None and err == "too_many_queued" + + +# ── every way a slot comes back (§5.1) ────────────────────────────────────── + +def test_closing_returns_the_slot(_=None): + s = _slots(node=1) + _open(s, "a") + s.close("a") + assert s.in_use(DOWNLOAD) == 0 + + +def test_losing_the_session_returns_everything_it_held(_=None): + """The primary reclaim, and the reason a lease is scoped to a connection: + a closed tab, a quit browser and a dropped network all arrive here, and + none of them needs a timer.""" + s = _slots(node=8, per_member=8) + _open(s, "a", session="doomed") + _open(s, "b", session="doomed") + _open(s, "c", session="other") + gone, _ = s.release_session("doomed") + assert sorted(x.tr for x in gone) == ["a", "b"] + assert s.in_use(DOWNLOAD) == 1 + + +def test_a_queued_lease_dies_with_its_session_too(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a", session="s1") + _open(s, "waiting", session="doomed") + s.release_session("doomed") + assert "waiting" not in s.leases + assert s.queues[DOWNLOAD] == [] + + +def test_a_grant_nobody_takes_up_is_passed_on(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a", now=0.0) + _open(s, "b", now=0.0) + ended, granted = s.sweep(now=GRANT_DEADLINE_SECS + 1) + assert [(x.tr, r) for x, r in ended] == [("a", REASON_NOT_TAKEN_UP)] + assert [x.tr for x in granted] == ["b"], "the slot was not passed on" + assert s.leases["a"].state == "queued", "the abandoned one goes to the tail" + + +def test_a_transfer_that_started_is_not_mistaken_for_an_abandoned_grant(_=None): + s = _slots(node=1) + _open(s, "a", now=0.0) + s.touch("a", now=1.0) + ended, _ = s.sweep(now=GRANT_DEADLINE_SECS + 2) + assert ended == [], "a transfer that is running was revoked" + + +def test_a_transfer_that_goes_quiet_is_reclaimed(_=None): + s = _slots(node=1) + _open(s, "a", now=0.0) + s.touch("a", now=1.0) + ended, _ = s.sweep(now=1.0 + IDLE_TIMEOUT_SECS + 1) + assert [(x.tr, r) for x, r in ended] == [("a", REASON_IDLE)] + assert "a" not in s.leases + + +def test_activity_keeps_a_slow_transfer_alive(_=None): + """A slow reader is not an absent one. The idle clock follows the lease's + own activity, not the wall since it started.""" + s = _slots(node=1) + _open(s, "a", now=0.0) + t = 0.0 + for _ in range(10): + t += IDLE_TIMEOUT_SECS - 1 + s.touch("a", now=t) + assert s.sweep(now=t)[0] == [] + assert "a" in s.leases + + +# ── idempotence, which is what makes a reconnect safe ─────────────────────── + +def test_reopening_the_same_transfer_does_not_charge_twice(_=None): + s = _slots(node=8, per_member=2) + first = _open(s, "a") + again = _open(s, "a") + assert again is first + assert s.in_use(DOWNLOAD) == 1 + + +def test_another_session_cannot_adopt_a_lease(_=None): + s = _slots() + _open(s, "a", session="mine") + lease, err = s.open(tr="a", kind=DOWNLOAD, session_key="theirs", + user_id="u1", group_id="g1") + assert lease is None and err == "not_your_transfer" + + +# ── caps changed live ─────────────────────────────────────────────────────── + +def test_raising_a_cap_starts_what_was_waiting(_=None): + s = _slots(node=1, per_member=8) + _open(s, "a") + _open(s, "b") + granted = s.set_caps(node={DOWNLOAD: 4}) + assert [x.tr for x in granted] == ["b"] + + +def test_lowering_a_cap_does_not_interrupt_anything(_=None): + s = _slots(node=4, per_member=4) + for tr in "abcd": + _open(s, tr) + s.set_caps(node={DOWNLOAD: 1}) + assert s.in_use(DOWNLOAD) == 4, "a running transfer was taken away" + assert _open(s, "e").state == "queued" + + +# ── the property that matters (§5.3) ──────────────────────────────────────── + +@pytest.mark.parametrize("seed", range(25)) +def test_the_counter_never_drifts(seed): + """ + Random open/close/drop/sweep/resize, checked after every single step. + + A leaked slot is a race, and a test that reasons about the happy path + agrees with a broken implementation by construction. Two invariants, both + of which a real leak breaks: what the pool says is in use is exactly the + set of granted leases, and no queue entry names a lease that no longer + exists — the second being how "waiting for ever behind a ghost" starts. + """ + rng = random.Random(seed) + s = _slots(node=rng.randint(1, 4), per_member=rng.randint(1, 3)) + sessions = [f"s{i}" for i in range(4)] + users = ["u1", "u2", "u3"] + live: list[str] = [] + now = 0.0 + counter = 0 + + for _ in range(400): + before = {k: s.in_use(k) for k in KINDS} + member_before = {(k, m): s.member_in_use(k, m) + for k in KINDS + for m in {x.member for x in s.leases.values()}} + now += rng.uniform(0.0, 40.0) + action = rng.choice( + ["open", "open", "open", "close", "touch", "drop", "sweep", "caps"]) + if action == "open": + counter += 1 + tr = f"t{counter}" + lease, err = s.open( + tr=tr, kind=rng.choice(KINDS), session_key=rng.choice(sessions), + user_id=rng.choice(users), group_id="g1", now=now) + if lease is not None: + live.append(tr) + elif action == "close" and live: + s.close(live.pop(rng.randrange(len(live))), now=now) + elif action == "touch" and live: + s.touch(rng.choice(live), now=now) + elif action == "drop": + s.release_session(rng.choice(sessions), now=now) + elif action == "sweep": + s.sweep(now=now) + elif action == "caps": + s.set_caps(node={rng.choice(KINDS): rng.randint(1, 5)}, now=now) + live = [tr for tr in live if tr in s.leases] + + for kind in KINDS: + granted = [x for x in s.leases.values() + if x.kind == kind and x.state == "granted"] + assert s.in_use(kind) == len(granted) + # Not `in_use <= cap`: lowering a cap never interrupts a transfer + # that is running, so the count legitimately sits above the new + # value until those finish. What must never happen is a *new* grant + # while the pool is at or over its cap -- so the count may fall or + # hold, and may only rise while there was room. + assert s.in_use(kind) <= max(s.caps[kind], before[kind]), ( + f"{kind}: {before[kind]} -> {s.in_use(kind)} granted with a cap " + f"of {s.caps[kind]} — a slot was handed out past the cap") + for tr in s.queues[kind]: + assert tr in s.leases, "a queue entry outlived its lease" + assert s.leases[tr].state == "queued" + for member in {x.member for x in granted}: + assert s.member_in_use(kind, member) <= max( + s.per_member[kind], member_before.get((kind, member), 0)) + + # And at the end: drop every session and nothing may be left holding + # anything. A slot that survives the last connection is a slot nothing can + # ever release. + for session in sessions: + s.release_session(session, now=now) + assert s.leases == {} + assert all(q == [] for q in s.queues.values()) + assert all(s.in_use(k) == 0 for k in KINDS) + + +# ── the cycle the node's own log showed ───────────────────────────────────── + +def test_a_grant_is_not_requeued_for_ever(_=None): + """ + A revoked grant went back in the queue, was granted again a millisecond + later because there was room, and was revoked again 30 s on. The node + logged the same two reclaims every 30 s for as long as it ran — minutes + after the transfers involved had finished. + + Three chances, then it is closed and the peer told, which is what ends the + cycle. `test_the_counter_never_drifts` could not see this: nothing drifted, + the same lease simply never left. + """ + s = _slots(node=4, per_member=4) + _open(s, "ghost", now=0.0) + now = 0.0 + reasons = [] + for _ in range(6): + now += GRANT_DEADLINE_SECS + 1 + ended, _granted = s.sweep(now=now) + reasons += [r for _, r in ended] + assert reasons.count(REASON_NOT_TAKEN_UP) == MAX_MISSED_GRANTS - 1 + assert reasons.count(REASON_ABANDONED) == 1 + assert "ghost" not in s.leases, "the lease is still cycling" + assert s.queues[DOWNLOAD] == [] + + +def test_a_transfer_that_is_running_is_never_revoked(_=None): + """ + The other half, and the one that mattered: nothing marked a lease used, so + `used` stayed False for a whole download and the sweeper revoked a grant + every 30 s while the file transferred at 20 MB/s. + """ + s = _slots(node=2, per_member=2) + _open(s, "live", now=0.0) + now = 0.0 + for _ in range(10): + now += GRANT_DEADLINE_SECS - 5 + assert s.touch("live", now=now), "a granted lease refused a touch" + ended, _granted = s.sweep(now=now) + assert ended == [], f"a running transfer was revoked: {ended}" + assert s.leases["live"].state == "granted" + + +def test_using_a_lease_forgives_its_earlier_misses(_=None): + """A slow start is not an abandoned one: a client that took two grants to + get going must not be closed on its third.""" + s = _slots(node=2, per_member=2) + _open(s, "slow", now=0.0) + s.sweep(now=GRANT_DEADLINE_SECS + 1) + s.sweep(now=2 * GRANT_DEADLINE_SECS + 2) + assert s.leases["slow"].missed_grants == 2 + s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3) + assert s.leases["slow"].missed_grants == 0 diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py new file mode 100644 index 0000000..db8c17a --- /dev/null +++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py @@ -0,0 +1,346 @@ +""" +Transfer leases over the session, rather than over `TransferSlots` alone. + +test_transfer_slots.py proves the decisions; this proves the seam. Both exist +because the seam is where this repo's defects have actually lived — a reply +routed by arrival order, a session popped from a dict without its work being +stopped, a slot released by a `finally` nobody reached. + +Three things can only be checked here: + + - the handlers answer under the right shape, and refuse another connection's + transfer id; + - **losing the connection gives everything back.** That is the primary + reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test + of the pool alone would never touch it; + - a slot freed by one peer is *announced* to the peer waiting on it. A grant + nobody hears about is precisely the "stuck at waiting" report the design + exists to prevent, and it would look correct in the pool. +""" + +import pytest + +# Every test drives a message handler, and in the node a message handler always +# runs inside the event loop: `_do_transfer_open` starts the sweeper task there. +# Calling these synchronously tested a situation that cannot happen and failed +# on "no current event loop" the moment the sweeper stopped being faked. +pytestmark = pytest.mark.asyncio + +from meshbay_common.protocol import MNP +from meshbay_node.transfers import DOWNLOAD, UPLOAD +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + + +class _Session(WebRTCPeerSession): + """A session with the DataChannel replaced by a list, and nothing else.""" + + def __init__(self, ctx, *, key, user, group="g1"): + self._ctx = ctx + self._registry_key = key + self._user_id = user + self._group_id = group + self.sent: list[dict] = [] + + def _send(self, msg): + self.sent.append(msg) + + def _spawn(self, coro): # pragma: no cover - not used by these tests + coro.close() + return None + + def last(self, mtype=MNP.TRANSFER_STATE): + return next(m for m in reversed(self.sent) if m.get("type") == mtype) + + +@pytest.fixture +def ctx(): + """A transport context, with the sweeper stopped on the way out. + + A task left running past the end of its test is a warning in the next one + and a hang in the worst case; the sweeper is started on demand by design, so + tearing it down is the test's job. + """ + c: dict = {"_peers": {}} + yield c + task = c.get("_transfer_sweeper") + if task is not None: + task.cancel() + + +def _join(ctx, key, user, group="g1") -> _Session: + s = _Session(ctx, key=key, user=user, group=group) + ctx["_peers"][key] = s + return s + + +async def test_a_granted_transfer_is_answered_as_granted(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10}) + reply = peer.last() + assert reply["state"] == "granted" + assert reply["tr"] == "t1" + assert reply["kind"] == DOWNLOAD + assert reply["used"] == 1 and reply["cap"] >= 1 + + +async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx): + peer = _join(ctx, "s1", "alice") + peer._slots().per_member[DOWNLOAD] = 1 + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_open({"tr": "t2"}) + peer._do_transfer_open({"tr": "t3"}) + assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"] + assert peer.sent[-1]["ahead"] == 1 + + +async def test_the_reply_carries_no_name_and_no_path(ctx): + """A lease holds neither, and `transfer_state` stays in clear — so this is + the message where a filename would quietly become metadata on the wire.""" + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv", + "path": "/srv/films"}) + assert set(peer.last()) <= { + "type", "v", "tr", "state", "kind", "used", "cap", "node_used", + "node_cap", "ahead", "reason"} + + +async def test_closing_frees_the_slot(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_close({"tr": "t1", "reason": "done"}) + assert peer.last()["state"] == "closed" + assert peer._slots().in_use(DOWNLOAD) == 0 + + +async def test_one_peer_cannot_close_anothers_transfer(ctx): + """A denial of service one random id away, otherwise.""" + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + alice._do_transfer_open({"tr": "t1"}) + bob._do_transfer_close({"tr": "t1"}) + assert bob.last("error")["code"] == "not_your_transfer" + assert "t1" in alice._slots().leases + + +async def test_one_peer_cannot_open_on_anothers_id(ctx): + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + alice._do_transfer_open({"tr": "t1"}) + bob._do_transfer_open({"tr": "t1"}) + assert bob.last("error")["code"] == "not_your_transfer" + + +async def test_a_transfer_with_no_id_is_refused(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"kind": DOWNLOAD}) + assert peer.last("error")["code"] == "bad_transfer_id" + + +# ── the reclaim that matters ──────────────────────────────────────────────── + +async def test_losing_the_connection_gives_everything_back(ctx): + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_open({"tr": "t2", "kind": UPLOAD}) + peer._release_transfers() + slots = peer._slots() + assert slots.leases == {} + assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0 + + +async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx): + """ + The seam this file exists for. In the pool, granting is correct; if the + grant is not pushed, the waiting client sits on "waiting" for ever with a + node that believes it is streaming — and every unit test still passes. + """ + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + alice._slots().caps[DOWNLOAD] = 1 + + alice._do_transfer_open({"tr": "a1"}) + bob._do_transfer_open({"tr": "b1"}) + assert bob.last()["state"] == "queued" + + alice._release_transfers() + assert bob.last()["state"] == "granted", ( + "bob was granted the slot and never told") + assert bob.last()["tr"] == "b1" + + +async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx): + """The pools are node-wide and `_peer_registry` is per group (finding H1), + so the peer to notify is not necessarily in the notifier's own registry.""" + groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}} + ctx = {"groups": groups} + alice = _Session(ctx, key="s1", user="alice", group="g1") + groups["g1"]["_peers"]["s1"] = alice + bob = _Session(ctx, key="s2", user="bob", group="g2") + groups["g2"]["_peers"]["s2"] = bob + + alice._slots().caps[DOWNLOAD] = 1 + alice._do_transfer_open({"tr": "a1"}) + bob._do_transfer_open({"tr": "b1"}) + assert bob.last()["state"] == "queued" + + alice._release_transfers() + assert bob.last()["state"] == "granted", ( + "a slot freed in one group never reached the peer waiting in another") + + +async def test_raising_the_cap_notifies_who_it_starts(ctx): + """`set_capacity` arrives from the loopback API, with no session behind it — + the grants it produces still have to be pushed.""" + from meshbay_node.transport.webrtc_server import WebRTCTransport + + transport = WebRTCTransport.__new__(WebRTCTransport) + transport._ctx = ctx + peer = _join(ctx, "s1", "alice") + peer._slots().caps[DOWNLOAD] = 1 + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_open({"tr": "t2"}) + assert peer.last()["state"] == "queued" + + transport.set_capacity(max_concurrent_downloads=4) + assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2" + + +async def test_the_operator_can_see_the_queue(ctx): + """`GET /api/transfers` is the answer to "was this peer ever queued", which + a log line cannot give when the symptom is that nothing is happening.""" + from meshbay_node import ops + + peer = _join(ctx, "s1", "alice") + peer._slots().caps[DOWNLOAD] = 1 + peer._do_transfer_open({"tr": "t1", "bytes": 5}) + peer._do_transfer_open({"tr": "t2", "bytes": 7}) + + class _T: + _ctx = ctx + + snapshot = await ops.list_transfers({"webrtc": _T()}) + assert snapshot["pools"][DOWNLOAD]["in_use"] == 1 + assert snapshot["pools"][DOWNLOAD]["queued"] == 1 + assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"} + assert all("name" not in x and "path" not in x for x in snapshot["leases"]) + + +async def test_asking_before_anything_has_transferred_is_not_an_error(): + from meshbay_node import ops + + class _T: + _ctx: dict = {} + + snapshot = await ops.list_transfers({"webrtc": _T()}) + assert snapshot["leases"] == [] + assert snapshot["pools"][DOWNLOAD]["in_use"] == 0 + + +@pytest.mark.asyncio +async def test_the_caps_shown_are_the_operators_before_anything_transfers(): + """ + `transfers set 2 2` answers "applied now"; `transfers show` said 0/8 — + because the no-pool branch reported the module defaults rather than what the + operator had just set. Found by running it against a real node. The previous + test asserted the defaults, so it agreed with the bug: an operator would + have read that as the hot-swap doing nothing all over again. + """ + from meshbay_node import ops + + class _T: + _ctx = {"max_concurrent_downloads": 2, "max_concurrent_uploads": 3} + + snapshot = await ops.list_transfers({"webrtc": _T()}) + assert snapshot["pools"][DOWNLOAD]["cap"] == 2 + assert snapshot["pools"][UPLOAD]["cap"] == 3 + + +# ── the sweeper's lifetime ────────────────────────────────────────────────── + +async def test_the_sweeper_outlives_the_session_that_started_it(ctx): + """ + It was started with `self._spawn`, which ties a task to one session's set — + so it was cancelled the moment that peer left, and every other peer's + abandoned lease stopped being reclaimed. Nothing else would have noticed: + the node simply fills up over days. + """ + import asyncio + + from meshbay_node.transport import webrtc_server as ws + + alice = _join(ctx, "s1", "alice") + bob = _join(ctx, "s2", "bob") + # The real _spawn, so the session genuinely owns what it starts. + alice._tasks = set() + alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice) + bob._tasks = set() + bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob) + + alice._do_transfer_open({"tr": "a1"}) + bob._do_transfer_open({"tr": "b1"}) + sweeper = ctx["_transfer_sweeper"] + assert sweeper is not None and not sweeper.done() + + # Alice leaves, exactly as shutdown_tasks does it. + alice._release_transfers() + for task in list(alice._tasks): + task.cancel() + await asyncio.gather(*alice._tasks, return_exceptions=True) + await asyncio.sleep(0) + + assert not sweeper.done(), ( + "the sweeper died with the session that happened to start it; bob's " + "lease would never be reclaimed") + sweeper.cancel() + + +async def test_the_sweeper_stops_when_the_last_lease_goes(ctx): + """An idle node must run no timer — the reason this is started on demand + rather than at boot.""" + import asyncio + + from meshbay_node.transport import webrtc_server as ws + + peer = _join(ctx, "s1", "alice") + peer._tasks = set() + peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer) + + original = ws.TRANSFER_SWEEP_SECS + ws.TRANSFER_SWEEP_SECS = 0.01 + try: + peer._do_transfer_open({"tr": "t1"}) + peer._do_transfer_close({"tr": "t1"}) + sweeper = ctx["_transfer_sweeper"] + await asyncio.wait_for(sweeper, timeout=2) + assert ctx.get("_transfer_sweeper") is None + finally: + ws.TRANSFER_SWEEP_SECS = original + + +async def test_a_chunk_request_keeps_its_lease_alive(ctx): + """ + The seam that cost an afternoon. `TransferSlots.touch` existed, was tested, + and **nothing ever called it**: the node ignored `tr` on `file_req` + entirely, so `used` stayed False for every download ever made and the + sweeper revoked each grant 30 s in, while the file was transferring. + + Neither side's tests could see it — the pool was correct, the handlers were + correct, and the call between them was missing. Only the node's own log + showed it, repeating the same reclaim every 30 s. + """ + peer = _join(ctx, "s1", "alice") + peer._do_transfer_open({"tr": "t1", "bytes": 1024}) + lease = peer._slots().leases["t1"] + assert lease.used is False + + # A chunk request for a file that does not exist still counts: what marks + # the lease is the peer asking, not the node succeeding. + peer._group_ctx()["index"] = None + try: + await peer._do_file_request({"file_id": "nope", "chunk_index": 0, + "tr": "t1"}) + except Exception: + pass + assert peer._slots().leases["t1"].used is True, ( + "a chunk request under this lease did not mark it alive; the node will " + "revoke the grant in 30 seconds") diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index c1a5287..284c461 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -1226,7 +1226,14 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di transport._ctx["roster"] = roster transport._ctx["has_admin_authority"] = True transport._ctx["groups"] = { - TEST_GROUP: {"gek": gek, "roots": shared_dir, "index": indexer.index}, + # A RootSet, like the transport two lines up and like the code under + # test expects: a group's content became several named roots (draft v6, + # change 1) and this one line kept passing the bare Path. The handshake + # died on `'PosixPath' object has no attribute 'describe'` and answered + # `error` instead of `handshake_ack`, which is a scaffolding that never + # followed the change, not a defect in the flow being tested. + TEST_GROUP: {"gek": gek, "roots": one_root(shared_dir), + "index": indexer.index}, } # `create_invite` registers the invitee as a hub member *before* writing the # invite, and fails the whole operation if it cannot: `/v1/groups/mine` diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py new file mode 100755 index 0000000..38e6cb6 --- /dev/null +++ b/packages/meshbay-node/tests/transfer_probe.py @@ -0,0 +1,596 @@ +#!/usr/bin/env python3 +""" +Ask a real node for more transfer slots than it has, and watch what it does. + +Everything about transfer slots has so far been proved against a `TransferSlots` +object and against sessions with a list where the DataChannel should be. Both +are worth having and neither has ever met a node. This speaks the same MNP over +the same WebRTC DataChannel as a browser, so what it measures is what a member +would get. + +Three questions, and only the first is about the cap: + + 1. **Is the cap real?** Open more transfers than it allows and count how many + come back granted. A node that grants everything is a node where none of + this does anything — which, until MNP 3.0 makes leases compulsory, is also + what an *old* client gets, so the number here is the difference between + "built" and "working". + 2. **Does the queue drain?** Close a granted transfer and see whether the + grant reaches whoever was waiting. The node can be perfectly right about + who deserves the slot and still never say so; the push is a separate thing + from the decision, and this is the only place both run. + 3. **Does a slot come back when a peer vanishes?** Drop the connection + without closing anything — a closed tab, a dead network — and ask a second + account whether the slot freed. That reclaim is a hook, not a timeout, so + it should be immediate. + + .venv/bin/python packages/meshbay-node/tests/transfer_probe.py --group <id> + --want 6 # ask for six slots at once + --keep-open # hold them, then look at `meshbay-node transfers` + --pull 3 # download three files to completion, under a lease + --pull 3 --parallel # …at the same time on one connection + +**Not collected by pytest** — the filename does not match `test_*.py`, which is +deliberate: this talks to a real hub with real credentials and takes minutes. +It lives here rather than in `QE/` so it survives, because `QE/` is not +versioned and this probe found several defects that no test in the suite could +reach: a cap that was never enforced, a queue that granted a slot and never said +so, and leases that outlived the session holding them. + +What it still needs from `QE/`, which stays out of the repo: + + - `QE/deploy/e2e.py` — the second implementation of the client, whose `Client` + speaks MNP over a real WebRTC DataChannel. Located at run time; the probe + says so plainly if it is missing rather than failing on an import. + - `QE/deploy/demo.env` — credentials. Never in the repo, by the same rule. + +Nothing here writes to the node: a lease is in-memory state that dies with the +connection, so the worst a failed run leaves behind is a slot the node reclaims +on its own. +""" + +import argparse +import asyncio +import sys +import uuid +from pathlib import Path + +# `e2e.py` is the MNP client this probe drives, and it lives in QE/, which is +# not versioned (credentials and test artefacts go there by convention). Found +# by walking up to the repo root rather than assumed to be a sibling, and its +# absence is explained rather than raised as an ImportError from four frames +# down. +_QE = Path(__file__).resolve().parents[3] / "QE" / "deploy" +if not (_QE / "e2e.py").exists(): + raise SystemExit( + f"{__file__.split('/')[-1]} needs QE/deploy/e2e.py, which is not in\n" + f"this checkout ({_QE} does not have it). QE/ is deliberately not\n" + f"versioned: it holds credentials and test artefacts. Copy it there, or\n" + f"run this from a machine that has one.") +sys.path.insert(0, str(_QE)) + +import httpx # noqa: E402 + +from e2e import Client, env # noqa: E402 + + +def _line(ok: bool, text: str) -> None: + print(f" [{'PASS' if ok else 'FAIL'}] {text}") + + +async def _open_transfer(client: Client, kind: str = "download", + nbytes: int = 1 << 30) -> tuple[str, dict]: + """One `transfer_open`, and the state for **that** transfer. + + Matched on `tr`, never on arrival order. `transfer_state` is also how a + close is acknowledged and how a grant is pushed minutes later, so "the next + one" is somebody else's answer as soon as more than one transfer is in + play — this probe read two stale `closed` acks as the replies to two opens + and reported the cap as broken. It is the same defect `req_id` exists for, + in the tool written to check the thing. + """ + tr = uuid.uuid4().hex + client.send({"type": "transfer_open", "v": "0.1", "tr": tr, + "kind": kind, "bytes": nbytes, "chunks": 1024}) + while True: + reply = await client.recv_type("transfer_state", timeout=15) + if reply.get("type") == "error" or reply.get("tr") == tr: + return tr, reply + + +async def pull(client, ack, count: int, parallel: bool = False) -> int: + """Download files to completion over MNP, under a lease, and say where they + stop. + + The browser is the hard place to look: a download that freezes near the end + there could be the service worker's backpressure, the DataChannel, the + node's own buffer wait, or the lease being revoked underneath it — and the + interface says the same thing for all four. This client holds no + SourceBuffer, no service worker and no iframe, so if a file arrives whole + the node and the transport are cleared and the browser is implicated. + """ + import time as _t + + # Asked for, not waited for: the node pushes an index when it changes, but + # a client that has just connected has to request one. Waiting for a push + # that may never come is a twenty-second timeout that says nothing. + client.send({"type": "index_sync", "v": "0.1"}) + index = await client.recv_type("index_sync", timeout=30) + entries = client.open_index(index).get("entries", []) + big = sorted([e for e in entries if e.get("size", 0) > 50 * 1024 * 1024], + key=lambda e: -e["size"])[:count] + if not big: + print("no file over 50 MB in this group to pull") + return 1 + + CHUNK = 1024 * 1024 + failures = 0 + + if parallel: + return await pull_together(client, big, CHUNK) + + for entry in big: + total_chunks = -(-entry["size"] // CHUNK) + tr, state = await _open_transfer(client, nbytes=entry["size"], + kind="download") + while state.get("state") == "queued": + state = await client.recv_type("transfer_state", timeout=120) + print(f"\n{entry['name'][:52]:<52} {entry['size'] / 1048576:8.1f} MB") + + got = 0 + started = _t.monotonic() + last_report = started + try: + for i in range(total_chunks): + client.send({"type": "file_req", "v": "0.1", + "file_id": entry["id"], "chunk_index": i, + "tr": tr}) + msg = await client.recv_type("file_chunk", timeout=90) + if msg.get("type") == "error": + raise RuntimeError(msg.get("detail", "refused")) + got += len(msg.get("ct") or b"") + if _t.monotonic() - last_report > 5: + last_report = _t.monotonic() + print(f" {got / 1048576:8.1f} MB chunk {i + 1}/{total_chunks}") + except Exception as exc: + pct = 100 * got / max(1, entry["size"]) + print(f" STOPPED at {got / 1048576:.1f} MB ({pct:.1f}%), " + f"chunk of {total_chunks}: {type(exc).__name__}: {exc}") + failures += 1 + client.send({"type": "transfer_close", "v": "0.1", "tr": tr, + "reason": "failed"}) + continue + + secs = _t.monotonic() - started + ok = got >= entry["size"] + print(f" {'COMPLETE' if ok else 'SHORT'} — {got / 1048576:.1f} MB in " + f"{secs:.0f}s ({got / 1048576 / max(secs, 1):.1f} MB/s)") + failures += not ok + client.send({"type": "transfer_close", "v": "0.1", "tr": tr, + "reason": "done"}) + + print() + print("every file arrived whole" if not failures + else f"{failures} file(s) did not arrive whole") + return 1 if failures else 0 + + +async def pull_together(client, entries, CHUNK) -> int: + """Every file at once, on one connection, interleaved. + + This is the shape a browser makes and the one a sequential pull cannot + reproduce: several downloads share a single DataChannel, each keeping a + window of chunk requests in flight, so the node's send buffer is under + pressure from all of them at once and every reply waits behind the others. + A download that arrives whole on its own can still stall here. + + Replies are matched by (file_id, chunk_index) rather than by arrival order, + because with several downloads in flight arrival order means nothing — + which is the same reason `req_id` exists. + """ + import time as _t + + WINDOW = 8 # PIPELINE_WINDOW in file-utils.js + state = {} + for e in entries: + tr, st = await _open_transfer(client, nbytes=e["size"], kind="download") + while st.get("state") == "queued": + st = await client.recv_type("transfer_state", timeout=180) + state[e["id"]] = {"entry": e, "tr": tr, "sent": 0, "got": 0, + "bytes": 0, "total": -(-e["size"] // CHUNK)} + print(f"{e['name'][:52]:<52} {e['size'] / 1048576:8.1f} MB") + + def fire(): + for st in state.values(): + while st["sent"] < st["total"] and st["sent"] - st["got"] < WINDOW: + client.send({"type": "file_req", "v": "0.1", + "file_id": st["entry"]["id"], + "chunk_index": st["sent"], "tr": st["tr"]}) + st["sent"] += 1 + + fire() + started = last = _t.monotonic() + stalled = None + while any(st["got"] < st["total"] for st in state.values()): + try: + msg = await client.recv_type("file_chunk", timeout=45) + except Exception as exc: + stalled = f"{type(exc).__name__}: {exc}" + break + if msg.get("type") == "error": + stalled = f"node refused: {msg.get('detail')}" + break + st = state.get(msg.get("file_id")) + if st is None: + continue + st["got"] += 1 + st["bytes"] += len(msg.get("ct") or b"") + fire() + if _t.monotonic() - last > 5: + last = _t.monotonic() + print(" " + " | ".join( + f"{s['entry']['name'][:14]:<14} {s['got']:>4}/{s['total']}" + for s in state.values())) + + print() + failures = 0 + for st in state.values(): + done = st["got"] >= st["total"] + failures += not done + print(f" {'COMPLETE' if done else 'STOPPED '} " + f"{st['entry']['name'][:44]:<44} " + f"{st['bytes'] / 1048576:8.1f} MB " + f"chunk {st['got']}/{st['total']}") + client.send({"type": "transfer_close", "v": "0.1", "tr": st["tr"], + "reason": "done" if done else "failed"}) + if stalled: + print(f"\n stalled after {_t.monotonic() - started:.0f}s: {stalled}") + print() + print("every file arrived whole" if not failures + else f"{failures} of {len(state)} did not arrive whole") + return 1 if failures else 0 + + +def _cli(*argv) -> str: + """Run `meshbay-node …` the way the operator does, and return its output.""" + import subprocess + out = subprocess.run(["meshbay-node", *argv], capture_output=True, + text=True, timeout=30) + if out.returncode != 0: + raise RuntimeError(f"meshbay-node {' '.join(argv)}: {out.stderr.strip()}") + return out.stdout + + +async def operator_checks(client, ack, group, node_id) -> int: + """The two things only the operator's side can answer. + + Both are about a promise made elsewhere: draft-v6 §2.11 says these settings + apply without a restart, and §5 of the transfer-slots plan says a lost peer + gives its slots back through a hook rather than a timeout. Neither can be + checked from inside the client, and both were wrong at some point today — + the live cap because the hot-swap wrote to an attribute that never existed, + and the operator's view because it reported the module defaults. + """ + import asyncio as _a + import json as _json + import re as _re + + failures = 0 + member_cap = int((ack.get("transfer_limits") or {}).get("download") or 0) + + # The queue has to be held by the *node* cap, not by this member's own. + # + # With both at 2, one account holding two transfers hits both at once, and + # raising the node-wide cap then correctly changes nothing — per-member is + # checked first, by design. An earlier version of this check set it up that + # way and reported the design working as a failure. So: node cap to 1, well + # under the member cap, and the third transfer is waiting on the machine. + was = _cli("transfers", "show") + prior = int(_re.search(r"download\s+\d+/(\d+)", was).group(1)) + _cli("transfers", "set", "1", "1") + + granted_tr, first = await _open_transfer(client) + tr_waiting, waiting = await _open_transfer(client) + ok = first.get("state") == "granted" and waiting.get("state") == "queued" + _line(ok, "with the node cap at 1, the second transfer waits on the machine " + f"rather than on this member's own cap of {member_cap}") + failures += not ok + + # ── 1. the operator can see it ──────────────────────────────────────── + shown = _cli("transfers", "show") + # The lease table only: "queued" also appears in each pool's summary line, + # and counting those reported three queues where there was one. + # The lease table only. The pool summary above it also says "download" and + # "0 queued", and counting those reported two queues where there was one — + # the parser broke when `transfers show` gained a per-group section, which + # is what a probe that reads a human-facing format signs up for. + lines = shown.splitlines() + head = next((i for i, ln in enumerate(lines) + if "transfer" in ln and "kind" in ln and "state" in ln), None) + rows = lines[head + 1:] if head is not None else [] + seen = sum(1 for ln in rows if " granted " in ln) + queued = sum(1 for ln in rows if " queued " in ln) + ok = seen == 1 and queued == 1 + _line(ok, f"`transfers show` lists {seen} granted and {queued} queued lease(s)") + failures += not ok + if not ok: + print(" the operator's only window into a stuck queue is wrong") + print(" " + shown.replace("\n", "\n ")) + + # ── 2. raising the cap live starts what was waiting ─────────────────── + _cli("transfers", "set", "4", "4") + try: + started = await _a.wait_for( + _wait_for_grant(client, tr_waiting), timeout=20) + except _a.TimeoutError: + started = False + _line(started, "raising the cap started the waiting transfer, with no " + "restart and no reconnection") + failures += not started + if not started: + print(" draft-v6 §2.11 promises this applies live; the setting " + "was accepted and nothing moved") + + # ── 2b. the *per-member* cap, which is a different door ─────────────── + # + # Checked separately because it is a different code path with a different + # front door, and only the node-wide one was covered: `set_capacity` pushed + # its grants and `ops.set_transfer_limits` computed them and forgot to send + # them. The pool was right, the peers were never told, and both transfers + # sat at "waiting" until the client's own watchdog re-asked a minute later. + # From a clean member: everything opened above is still held, and a check + # about "how many may one person run" cannot start with that person already + # holding several. An earlier version did and measured nothing. + for tr in [granted_tr, tr_waiting]: + client.send({"type": "transfer_close", "v": "0.1", "tr": tr, + "reason": "done"}) + # Waited for, not slept through: a close is a round trip, and measuring + # "how many may one person run" against a member who still holds two is + # measuring nothing. The operator's own view is the thing to wait on, + # because it is what the next assertion reads. + for _ in range(40): + if "nothing transferring" in _cli("transfers", "show"): + break + await _a.sleep(0.25) + else: + _line(False, "the member's earlier transfers never closed") + failures += 1 + _cli("transfers", "set", "8", "8") # node-wide out of the way + _cli("transfers", "per-member", "1", "1", "--group", group["id"]) + tr_first, first_held = await _open_transfer(client) + tr_c, held = await _open_transfer(client) + ok = first_held.get("state") == "granted" and held.get("state") == "queued" + _line(ok, "with the per-member cap at 1, a second transfer waits on it") + failures += not ok + if not ok: + print(f" first={first_held.get('state')} " + f"(used {first_held.get('used')}/{first_held.get('cap')}), " + f"second={held.get('state')} " + f"(used {held.get('used')}/{held.get('cap')})") + + _cli("transfers", "per-member", "4", "4", "--group", group["id"]) + try: + moved = await _a.wait_for(_wait_for_grant(client, tr_c), timeout=20) + except _a.TimeoutError: + moved = False + _line(moved, "raising the per-member cap started what was waiting on it") + failures += not moved + if not moved: + print(" the pool granted it and nobody told the peer — it sits " + "at 'waiting' until its own watchdog re-asks") + + # ── 3. a vanished peer's slots are back before anyone asks ──────────── + await client.close() + await _a.sleep(1.0) + after = _cli("transfers", "show") + ok = "nothing transferring" in after + _line(ok, "every slot came back when the peer vanished, with no timeout") + failures += not ok + if not ok: + print(" " + after.replace("\n", "\n ")) + + # Put the operator's cap back where it was found — not at a default, at + # whatever this node was running before the probe touched it. + _cli("transfers", "set", str(prior), str(prior)) + _cli("transfers", "per-member", str(member_cap), str(member_cap), + "--group", group["id"]) + print(f"\n (node cap restored to {prior}, per-member to {member_cap})") + + print() + print("all operator checks passed" if not failures + else f"{failures} operator check(s) failed") + return 1 if failures else 0 + + +async def _wait_for_grant(client, tr: str) -> bool: + """The node pushes the grant; nothing here polls for it. + + A grant that is decided and never sent is the "stuck at waiting" report the + whole design exists to prevent, and it looks perfectly correct in the pool. + """ + while True: + msg = await client.recv_type("transfer_state", timeout=30) + if msg.get("tr") == tr and msg.get("state") == "granted": + return True + + +async def probe(args) -> int: + cfg = env() + hub = cfg["HUB_URL"] + failures = 0 + + async with httpx.AsyncClient(timeout=30) as http: + alice = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"]) + await alice.login(http) + + # Which node is serving this group, resolved the way stream_probe.py + # does it. `connect()` needs the node id as well as the group: the hub + # relays signalling to one node, and a group may be hosted by more than + # one. + groups = (await http.get(f"{hub}/v1/groups/mine", + headers=alice.auth)).json()["groups"] + wanted = [g for g in groups + if g["id"] == args.group + or g["id"].startswith(args.group) + or args.group.lower() in g["name"].lower()] + if not wanted: + print(f"no group of yours matches {args.group!r}") + return 1 + group = wanted[0] + nodes = (await http.get(f"{hub}/v1/groups/{group['id']}/nodes", + headers=alice.auth)).json()["nodes"] + if not nodes: + print(f"no node online for {group['name']} — start it and retry") + return 1 + node_id = nodes[0]["node_id"] + print(f"group : {group['name']} ({group['id'][:8]})") + print(f"node : {node_id[:12]}\n") + + ack = await alice.connect(http, group["id"], node_id) + + limits = ack.get("transfer_limits") + if limits is None: + print("This node does not hand out transfer slots — it predates " + "them, or the handshake ack lost the field. Nothing below " + "can be measured.") + await alice.close() + return 1 + cap = int(limits.get("download") or 0) + print(f"node reports this member may run {cap} download(s) at once\n") + + if args.operator: + return await operator_checks(alice, ack, group, node_id) + + if args.pull: + return await pull(alice, ack, args.pull, args.parallel) + + # ── 1. is the cap real ──────────────────────────────────────────── + # + # The baseline first. A node that is already serving somebody grants + # this probe fewer slots than its cap, entirely correctly — and an + # earlier version reported that as two failures, which is a probe + # lying about a node that was right. Seen for real: a previous run of + # this script had crashed before closing, and its leases were still + # held. So the first reply is read for what the node says is already + # in use, and the run stops rather than measuring against a moving + # floor. + want = args.want or (cap + 2) + opened = [await _open_transfer(alice)] + first = opened[0][1] + if first.get("state") != "granted" or first.get("used", 1) != 1: + print(f"this node is not idle: it reports {first.get('used')} of " + f"{first.get('cap')} slots already used by this member, and " + f"{first.get('node_used')} of {first.get('node_cap')} " + f"node-wide.\nWait for it to settle (or `meshbay-node " + f"transfers show` to see what is holding them) and run again " + f"— the cap cannot be measured against a moving floor.") + await alice.close() + return 1 + for _ in range(want - 1): + opened.append(await _open_transfer(alice)) + granted = [r for _, r in opened if r.get("state") == "granted"] + queued = [r for _, r in opened if r.get("state") == "queued"] + print(f"asked for {want}: {len(granted)} granted, {len(queued)} queued") + ok = len(granted) == cap and len(queued) == want - cap + _line(ok, f"the cap is enforced ({len(granted)} granted against a cap " + f"of {cap})") + failures += not ok + + if queued: + positions = [r.get("ahead") for r in queued] + ok = positions == sorted(positions) and positions[0] == 0 + _line(ok, f"queue positions are handed out in order: {positions}") + failures += not ok + + if args.keep_open: + print("\nholding them. Look at the node with:\n" + " meshbay-node transfers show\n" + "Ctrl-C when done — every slot is released by the " + "disconnection alone.") + try: + await asyncio.Event().wait() + except (KeyboardInterrupt, asyncio.CancelledError): + pass + await alice.close() + return 0 + + # ── 2. does the queue drain ─────────────────────────────────────── + if queued: + first_tr = opened[0][0] + alice.send({"type": "transfer_close", "v": "0.1", "tr": first_tr, + "reason": "done"}) + # Two messages come back: the close, and the grant it produced. + # Which order is not promised, so both are collected. + seen = [] + for _ in range(2): + try: + seen.append(await alice.recv_type("transfer_state", + timeout=15)) + except asyncio.TimeoutError: + break + promoted = [m for m in seen if m.get("state") == "granted"] + ok = bool(promoted) + _line(ok, "closing a transfer granted the slot to the next in queue") + failures += not ok + if not ok: + print(" the node decided correctly and never said so — " + "the client would sit at 'waiting' for ever") + + # ── 3. does a vanished peer give its slots back ─────────────────── + # + # No second account needed, and that is not a compromise: a lease + # belongs to the *connection*, so a reconnecting client is a new session + # to the node. If the old one's leases were not released they still + # count against this member's cap, and the reconnect finds nothing free + # — which makes this the same check, on any group, without depending on + # who else happens to be a member. + # + # Dropped without closing anything: a shut tab, a dead network. The + # reclaim is a hook on the connection, not a timeout, so it should be + # immediate rather than two minutes away. + await alice.close() + await asyncio.sleep(1.5) + + again = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"]) + await again.login(http) + await again.connect(http, group["id"], node_id) + _, reborn = await _open_transfer(again) + ok = reborn.get("state") == "granted" and reborn.get("used") == 1 + _line(ok, "the slots came back when the peer vanished without closing") + failures += not ok + if not ok: + print(f" the node still counts {reborn.get('used')} of " + f"{reborn.get('cap')} against this member — the old session's " + f"leases outlived it, and only the idle sweep will free them") + await again.close() + + print() + if failures: + print(f"{failures} check(s) failed — do not make leases compulsory yet") + else: + print("all checks passed") + return 1 if failures else 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--group", required=True, help="group id to connect to") + ap.add_argument("--want", type=int, default=0, + help="how many transfers to open at once (default: cap + 2)") + ap.add_argument("--keep-open", action="store_true", + help="hold the transfers so the node can be inspected") + ap.add_argument("--pull", type=int, default=0, metavar="N", + help="actually download N files to completion, under a " + "lease, and report where they stop") + ap.add_argument("--operator", action="store_true", + help="the two checks that need the operator's CLI: a cap " + "raised live starts what was waiting, and a vanished " + "peer's slots are back before anyone asks") + ap.add_argument("--parallel", action="store_true", + help="pull them at the same time on one connection, the " + "way a browser does — which is when it goes wrong") + return asyncio.run(probe(ap.parse_args())) + + +if __name__ == "__main__": + raise SystemExit(main()) |