Merge branch 'main' of meshbay.org:meshbay

3 files changed, 153 insertions, 1 deletions

diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index c618ee4..47f82b3 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -79,6 +79,8 @@ class FakeNode:
         self.members: dict[str, Identity] = {}   # sender → the device it signs with
         self.unresponsive = False            # pings go unanswered, as a dead link's
         self.admin_ops: dict[str, dict] = {}
+        self.channels: list[FakeChannel] = []
+        self.index_syncs = 0                 # full indexes asked for
 
     def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
         fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest()
@@ -92,9 +94,26 @@ class FakeNode:
         return entry
 
     def session(self) -> GroupSession:
-        return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek,
+        ch = FakeChannel(self)
+        self.channels.append(ch)
+        return GroupSession(None, ch, GROUP_ID, "node-1", NODE_PK, self.gek,
                             dict(self.ack), self.identity, self.nonce, self.role)
 
+    def push_delta(self, *, additions=(), updates=(), deletions=()) -> None:
+        """An `index_delta` to every session, sealed as the node seals one. The
+        entries are applied here first, so a later `index_sync` agrees."""
+        for e in (*additions, *updates):
+            self.files[e["id"]] = e
+        for fid in deletions:
+            self.files.pop(fid, None)
+        wire = lambda es: [{k: v for k, v in e.items() if k != "data"} for e in es]  # noqa: E731
+        payload = {"base_version": 1, "version": 1, "additions": wire(additions),
+                   "deletions": list(deletions), "updates": wire(updates)}
+        msg = {"type": "index_delta", "group_id": GROUP_ID,
+               **seal(self.gek, PURPOSE_INDEX, "index_delta", GROUP_ID, payload)}
+        for ch in self.channels:
+            ch._emit(msg)
+
     def list_device(self, sender: str, device: Identity, *,
                     countersigned_by: Identity | None = None, proof: bool = True) -> None:
         """A roster row for `sender`'s device: a root, or countersigned by another
@@ -189,6 +208,7 @@ class FakeChannel:
 
     def _index_sync(self, msg):
         n = self.node
+        n.index_syncs += 1
         payload = {"version": 1, "roots": n.roots, "dirs": sorted(n.dirs),
                    "entries": [{k: v for k, v in e.items() if k != "data"}
                                for e in n.files.values()]}
diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py
index 36b7f47..9960533 100644
--- a/packages/meshbay-cli/tests/test_interactive.py
+++ b/packages/meshbay-cli/tests/test_interactive.py
@@ -444,3 +444,46 @@ async def test_put_sends_a_file_into_the_chat_folder_and_get_saves_one_here(
         await say("/get ../share/x")
         await _until(pilot, lambda: "name a file of the chat's folder" in _said(room))
         await pilot.press("escape")
+
+
+async def test_index_updates_are_applied_not_asked_for_again(world, monkeypatch):
+    """
+    A library being enriched pushes an `index_delta` about once a second, each
+    one filling in a file's fields. The panes used to answer every one with the
+    whole index — 67 full indexes in the 80 s of one download on a real node,
+    on the event loop the download was being received on. Updates are applied
+    where they land; only a delta that adds or removes a file asks again, once
+    for a burst.
+    """
+    from meshbay_cli.tui import screens
+    monkeypatch.setattr(screens, "REFRESH_AFTER_DELTA_S", 0.2)
+    hub, node, work = world
+    film = node.add("share", "film.mkv", b"F" * 5000)
+    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
+                                              "--password", PASSPHRASE, "login"]) == 0
+    app = _app()
+    async with app.run_test(size=(140, 40)) as pilot:
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        remote = app.screen.query_one(RemotePane)
+        await _open(pilot, remote, f"Family@{USERNAME}/")
+        await _open(pilot, remote, "share/")
+        await _until(pilot, lambda: "film.mkv" in _names(remote))
+        asked = node.index_syncs
+
+        for i in range(20):
+            node.push_delta(updates=[{**film, "duration": 100 + i}])
+        await pilot.pause(0.6)
+        assert node.index_syncs == asked, "an update must not fetch the whole index"
+        tree = app.screen._trees["g1"]
+        assert tree.file("share/film.mkv")["duration"] == 119
+
+        extra = [node.add("share", f"new{i}.txt", b"n%d" % i) for i in range(3)]
+        node.files = {k: v for k, v in node.files.items() if v not in extra}
+        for e in extra:
+            node.push_delta(additions=[e])
+        await _until(pilot, lambda: "new2.txt" in _names(remote))
+        await pilot.pause(0.6)
+        assert node.index_syncs == asked + 1, "a burst of additions asks again once"
+
+        node.push_delta(deletions=[film["id"]])
+        await _until(pilot, lambda: "film.mkv" not in _names(remote))
diff --git a/packages/meshbay-cli/tests/test_sctp_fix.py b/packages/meshbay-cli/tests/test_sctp_fix.py
new file mode 100644
index 0000000..04638db
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_sctp_fix.py
@@ -0,0 +1,89 @@
+"""The client's copy of the node's aiortc fixes (client/sctp_fix.py).
+
+The copy exists because neither package may depend on the other; it is held to
+the node's by comparing the code itself, docstrings aside, and the two effects
+that set a download's speed here are replayed against aiortc.
+"""
+
+import ast
+import asyncio
+from pathlib import Path
+
+import pytest
+from aiortc.rtcsctptransport import (
+    SCTP_DATA_FIRST_FRAG,
+    SCTP_DATA_LAST_FRAG,
+    DataChunk,
+    InboundStream,
+    RTCSctpTransport,
+    serialize_packet,
+)
+from meshbay_cli.client import sctp_fix
+
+NODE_COPY = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src" / "meshbay_node"
+             / "transport" / "webrtc" / "sctp_fix.py")
+
+
+def _code(path: Path) -> str:
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    tree.body = tree.body[1:]          # the module docstring
+    return ast.dump(tree)
+
+
+@pytest.mark.skipif(not NODE_COPY.exists(), reason="not in the source tree")
+def test_the_code_is_the_nodes():
+    assert _code(Path(sctp_fix.__file__)) == _code(NODE_COPY)
+
+
+def _chunks(n_fragments: int, count: int, tsn: int = 1000) -> list[DataChunk]:
+    out = []
+    for seq in range(count):
+        for i in range(n_fragments):
+            c = DataChunk()
+            c.flags = ((SCTP_DATA_FIRST_FRAG if i == 0 else 0)
+                       | (SCTP_DATA_LAST_FRAG if i == n_fragments - 1 else 0))
+            c.tsn, c.stream_id, c.stream_seq, c.protocol = tsn, 1, seq, 53
+            c.user_data = b"x"
+            tsn += 1
+            out.append(c)
+    return out
+
+
+def test_a_megabyte_is_walked_once(monkeypatch):
+    walks = []
+    original = sctp_fix._pop_messages
+    monkeypatch.setattr(sctp_fix, "_pop_messages",
+                        lambda s: (walks.append(1), original(s))[1])
+    s = InboundStream()
+    got = 0
+    for c in _chunks(870, 3):
+        s.add_chunk(c)
+        got += len(list(s.pop_messages()))
+    assert got == 3 and len(walks) == 3
+
+
+def test_every_second_packet_is_acknowledged():
+    class _Dtls:
+        state = "connected"
+
+        def _register_data_receiver(self, *a):
+            pass
+
+    async def go():
+        s = RTCSctpTransport(_Dtls())
+        s._association_state = s.State.ESTABLISHED
+        s._last_received_tsn = 999
+        sacks = []
+
+        async def send_sack():
+            sacks.append(s._last_received_tsn)
+            s._sack_needed = False
+
+        async def deliver(*_a):
+            pass
+
+        s._send_sack, s._receive = send_sack, deliver
+        for c in _chunks(1, 6):
+            await s._handle_data(serialize_packet(1, 1, s._local_verification_tag, c))
+        return sacks
+    assert asyncio.run(go()) == [1001, 1003, 1005]