1 files changed, 97 insertions, 1 deletions
diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py
index 206b77b..2811836 100644
--- a/packages/meshbay-node/tests/test_disk_io_off_loop.py
+++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py
@@ -29,12 +29,15 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from meshbay_common.crypto import generate_gek
 from meshbay_common.protocol import MNP
 from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
+from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
-from meshbay_node.roots import RootSet
+from meshbay_node.roots import Root, RootSet
 from meshbay_node.transfers import LeaselessReads
 from meshbay_node.transport import webrtc_server
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
+from conftest import one_root, sealed_upload
+
 GROUP = "g" * 32
 
 # Long enough that a blocked loop is unmistakable, short enough to keep the
@@ -222,3 +225,96 @@ def test_no_handler_resolves_a_path_on_the_loop():
     assert not stray, (
         f"{len(stray)} call(s) to entry_abs_path outside _locate: "
         f"resolve a path through `off_disk(roots, _locate, ...)` instead")
+
+
+async def test_the_availability_poll_does_not_stop_the_loop(tmp_path, monkeypatch):
+    """
+    The poll is the one that runs whether anybody asked for anything.
+
+    `refresh_availability` stats every root, and reconcile calls it on a timer.
+    On the loop, a node with a sleeping disk stalls once per tick for as long as
+    the spin-up takes — and the stat is also what keeps waking the disk, so the
+    node pays for a library nobody is reading.
+    """
+    root = tmp_path / "films"
+    root.mkdir()
+    (root / "clip.bin").write_bytes(CONTENT)
+    roots = RootSet.build([{"path": str(root), "name": "films"}])
+    monkeypatch.setattr(Root, "is_live", _slow(Root.is_live))
+
+    idx = DirectoryIndexer(roots=roots, group_id=GROUP,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None)
+    try:
+        with _Ticker() as ticker:
+            await idx.initial_scan()
+    finally:
+        await idx.stop()
+        roots.close_io()
+
+    assert ticker.ticks > SLOW_S / TICK_S / 2, (
+        f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s poll")
+
+
+def _upload_session(tmp_path):
+    """One connection into a group with one writable root, as a node has."""
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    ctx = {"roots": one_root(shared),
+           "index": GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()),
+           "gek": generate_gek()}
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = GROUP
+    session._user_id = "user-1"
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session, shared
+
+
+async def test_a_slow_upload_write_does_not_stop_the_loop(tmp_path, monkeypatch):
+    session, shared = _upload_session(tmp_path)
+    monkeypatch.setattr(webrtc_server, "_append_chunk",
+                        _slow(webrtc_server._append_chunk))
+
+    with _Ticker() as ticker:
+        await session._do_file_upload(sealed_upload(
+            session, filename="clip.bin", data=CONTENT))
+
+    assert ticker.ticks > SLOW_S / TICK_S / 2, (
+        f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s write")
+    assert (shared / "clip.bin").read_bytes() == CONTENT
+
+
+async def test_chunks_of_one_upload_keep_their_order_under_a_slow_disk(tmp_path,
+                                                                      monkeypatch):
+    """
+    The rule that used to hold for free.
+
+    `chunk_index != state.next_index` is refused, and nothing could come between
+    that check and the `advance` answering it while the handler was synchronous.
+    Awaiting the write opens the gap: chunk 1 arriving while chunk 0 is still in
+    the disk thread reads a position that has not moved yet and is refused as
+    out of order — an upload that fails on a slow disk and nowhere else. The
+    lock closes it, and the disk made slow here is what makes the gap wide
+    enough to fall into.
+    """
+    session, shared = _upload_session(tmp_path)
+    pieces = [b"first-", b"second-", b"third-", b"fourth"]
+
+    monkeypatch.setattr(webrtc_server, "_append_chunk",
+                        _slow(webrtc_server._append_chunk))
+
+    # Fired together and in order, which is what the dispatcher does: it creates
+    # one task per message as it arrives.
+    await asyncio.gather(*[
+        session._do_file_upload(sealed_upload(
+            session, filename="clip.bin", data=piece,
+            chunk_index=i, total_chunks=len(pieces)))
+        for i, piece in enumerate(pieces)
+    ])
+
+    refusals = [m for m in session.sent if m.get("type") == "error"]
+    assert not refusals, f"a chunk was refused: {refusals}"
+    assert (shared / "clip.bin").read_bytes() == b"".join(pieces)