diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_hot_reload_survives_client_close.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_hot_reload_survives_client_close.py | 329 |
1 files changed, 329 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py new file mode 100644 index 0000000..7cb74cb --- /dev/null +++ b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py @@ -0,0 +1,329 @@ +""" +Adding a group (Create Group wizard, or "add a directory" to an existing +one) fires `_reload_config()` without awaiting it (`asyncio.ensure_future`, +ui/app.py) — the request handler, and whatever browser tab triggered it, +return immediately. This is deliberate: the initial scan behind it can take +a very long time (measured at 23 minutes for a 114 GB library on a slow +disk), and none of that work belongs to the HTTP request or the WebRTC +session that happened to start it. + +This test proves the scan is genuinely independent of its caller: it starts +the reload the same way the real endpoint does — schedules it and does not +await it, standing in for "the browser tab that made the call was closed" — +then does something else, and only afterwards checks that the reload +finished and the new group became available on its own. +""" + +import asyncio +import base64 +import os + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from unittest.mock import AsyncMock, MagicMock, patch + +from meshbay_node import ops +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +import meshbay_node.indexer.indexer as indexer_mod + + +def _free_port() -> int: + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _toml(data_dir, first_group_dir, second_group_id=None, second_group_dir=None) -> str: + # data_dir MUST come before any [section] header — TOML has no notion of + # "back to top-level" once a table is open, so a bare `key = value` line + # placed after [node] becomes node.data_dir, not the top-level data_dir + # load_config() actually reads. Silently falls back to the real default + # (~/.local/share/meshbay) instead of erroring, which is exactly how this + # test once ran a whole daemon — including _shutdown()'s unlink of + # ui-token — against the developer's real, already-running node. + text = f""" +data_dir = "{data_dir}" + +[hub] +url = "http://localhost:9999" +username = "testuser" + +[node] +quic_port = {_free_port()} +ui_port = {_free_port()} + +[[groups]] +id = "{"a" * 32}" +name = "first" +shared_dir = "{first_group_dir}" +visibility = "private" +""" + if second_group_id: + text += f""" +[[groups]] +id = "{second_group_id}" +name = "slow-new-group" +shared_dir = "{second_group_dir}" +visibility = "private" +""" + return text + + +def _mock_keystore_keys(sk_ed): + sk_x = X25519PrivateKey.generate() + pk_x_raw = sk_x.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + mock_keys = MagicMock() + mock_keys.sk_ed25519 = sk_ed + mock_keys.pk_ed25519_b64 = "test" + mock_keys.sk_x25519 = sk_x + mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode() + return mock_keys + + +@pytest.mark.asyncio +async def test_hot_loaded_group_finishes_scanning_without_anyone_awaiting_the_reload( + tmp_path): + first_dir = tmp_path / "first" + first_dir.mkdir() + (first_dir / "readme.txt").write_bytes(b"hello") + + second_dir = tmp_path / "second" + second_dir.mkdir() + for i in range(3): + (second_dir / f"file{i}.bin").write_bytes(os.urandom(64)) + second_group_id = "b" * 32 + + data_dir = tmp_path / "data" + config_path = tmp_path / "node.toml" + config_path.write_text(_toml(data_dir, first_dir)) + + from meshbay_node.config import load_config + daemon = NodeDaemon(load_config(config_path), config_path=config_path) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = _mock_keystore_keys(sk_node) + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = sk_node.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + # Slow the second group's hashing down (a stand-in for a large/slow + # library) so there is a real window in which "nobody is awaiting this" + # actually means something, without needing a genuinely huge file. + real_scan_file = indexer_mod._scan_file + + def slow_scan_file(root, path): + import time + time.sleep(0.15) + return real_scan_file(root, path) + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch.object(indexer_mod, "_scan_file", slow_scan_file): + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=15) + except (asyncio.TimeoutError, Exception): + pass + + run_task = asyncio.create_task(run_daemon()) + try: + for _ in range(50): + if daemon._state.get("status") == "running": + break + await asyncio.sleep(0.05) + assert daemon._state["status"] == "running" + assert second_group_id not in daemon._state.get("groups_ctx", {}) + + # Add the second group to the config on disk, the way the wizard's + # attach + /api/reload would leave it, then fire the reload exactly + # as ui/app.py does: scheduled, NOT awaited. + config_path.write_text(_toml(data_dir, first_dir, + second_group_id, second_dir)) + reload_task = asyncio.ensure_future(daemon._reload_config()) + + # Stand in for "the browser tab is gone": do something completely + # unrelated to the reload, and explicitly do not await it here. + await asyncio.sleep(0.01) + assert second_group_id not in daemon._state.get("groups_ctx", {}), \ + "the scan (3 files x 0.15s) cannot have finished yet" + + # Only now catch up with the background work, from a place that + # has no relationship to whatever originally triggered it. + await asyncio.wait_for(reload_task, timeout=5) + + assert second_group_id in daemon._state["groups_ctx"], \ + "the new group must be usable once its scan finishes, " \ + "regardless of whether anything was still watching the reload" + new_indexer = daemon._state["indexers"][second_group_id] + assert new_indexer.index.count == 3 + assert new_indexer.progress.scanning is False + finally: + shutdown_event.set() + await daemon._shutdown() + run_task.cancel() + try: + await run_task + except (asyncio.CancelledError, Exception): + pass + + +@pytest.mark.asyncio +async def test_group_scoped_ops_404_until_listed_then_succeed(tmp_path): + """ + The wizard's own sequence, reproduced against the real ops layer: attach + a brand-new group, fire the reload the way /api/reload now does + (ops.start_reload — scheduled, not awaited), and hit the group-scoped + calls that come right after in the UI (add a root, init the GEK) while + the scan is still running. + + Found live: "Attaching to node" no longer times out (ops.start_reload + returns immediately), but the very next wizard step then failed with + "Group not configured on this node" / "Group not hosted on this node" — + the group is not in daemon._state["config"].groups or ["groups_ctx"] + until _reload_config_inner() finishes, scan included, which is *after* + ops.start_reload has already returned. This locks in both halves: the + 404 while the scan runs, and success once ops.list_groups() actually + lists the group — the exact condition the wizard's own wait + (platform.waitForGroupHosted, app.js) polls for. + """ + first_dir = tmp_path / "first" + first_dir.mkdir() + (first_dir / "readme.txt").write_bytes(b"hello") + + second_dir = tmp_path / "second" + second_dir.mkdir() + for i in range(3): + (second_dir / f"file{i}.bin").write_bytes(os.urandom(64)) + second_group_id = "c" * 32 + extra_root_dir = tmp_path / "extra_root" + extra_root_dir.mkdir() + + data_dir = tmp_path / "data2" + config_path = tmp_path / "node2.toml" + config_path.write_text(_toml(data_dir, first_dir)) + + from meshbay_node.config import load_config + daemon = NodeDaemon(load_config(config_path), config_path=config_path) + + sk_node = Ed25519PrivateKey.generate() + mock_keys = _mock_keystore_keys(sk_node) + mock_session = MagicMock() + mock_session.node_id = "node123" + mock_session.user_id = "user123" + mock_session.hub_pk_pem = sk_node.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + real_scan_file = indexer_mod._scan_file + + def slow_scan_file(root, path): + import time + time.sleep(0.15) + return real_scan_file(root, path) + + with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ + patch("meshbay_node.daemon.HubClient") as MockHub, \ + patch.object(indexer_mod, "_scan_file", slow_scan_file): + + hub_instance = AsyncMock() + hub_instance.startup = AsyncMock(return_value=mock_session) + hub_instance.send_ws = AsyncMock() + hub_instance._ws = None + hub_instance.close = AsyncMock() + hub_instance.__aenter__ = AsyncMock(return_value=hub_instance) + hub_instance.__aexit__ = AsyncMock(return_value=False) + MockHub.return_value = hub_instance + + shutdown_event = asyncio.Event() + + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + hub_instance.maintain_ws = mock_maintain_ws + + async def run_daemon(): + with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=15) + except (asyncio.TimeoutError, Exception): + pass + + run_task = asyncio.create_task(run_daemon()) + try: + for _ in range(50): + if daemon._state.get("status") == "running": + break + await asyncio.sleep(0.05) + assert daemon._state["status"] == "running" + + # The same file write ops.attach_group does (a raw text append), + # then the same fire-and-forget reload /api/reload now does. + config_path.write_text(_toml(data_dir, first_dir, + second_group_id, second_dir)) + reload_task = asyncio.ensure_future(ops.start_reload(daemon._state)) + + await asyncio.sleep(0.01) + listing = await ops.list_groups(daemon._state) + assert second_group_id not in [g["id"] for g in listing["groups"]], \ + "the scan (3 files x 0.15s) cannot have finished this fast" + + # Exactly the wizard's next two steps, hit mid-scan. + with pytest.raises(ops.OpError) as add_root_exc: + await ops.add_root(daemon._state, second_group_id, + str(extra_root_dir)) + assert add_root_exc.value.status == 404 + + with pytest.raises(ops.OpError) as gek_exc: + await ops.set_gek(daemon._state, second_group_id) + assert gek_exc.value.status == 404 + + # Now wait the way platform.waitForGroupHosted (app.js) does: + # poll list_groups(), not index-status, until the group is + # actually there. + for _ in range(100): + listing = await ops.list_groups(daemon._state) + if second_group_id in [g["id"] for g in listing["groups"]]: + break + await asyncio.sleep(0.05) + else: + pytest.fail("group never appeared in list_groups()") + + await asyncio.wait_for(reload_task, timeout=5) + + # Both calls that 404'd above must now succeed. + add_result = await ops.add_root(daemon._state, second_group_id, + str(extra_root_dir)) + assert add_result["status"] == "added" + + gek_result = await ops.set_gek(daemon._state, second_group_id) + assert gek_result["status"] == "ok" + finally: + shutdown_event.set() + await daemon._shutdown() + run_task.cancel() + try: + await run_task + except (asyncio.CancelledError, Exception): + pass |