perf: linear SCTP reassembly, one SACK every second packet, 4 MB buffer in the CLI

aiortc walked a message's fragments after each one arrived (n^2/2 steps
for a 1 MiB chunk) and acknowledged every packet. Measured on loopback,
400 MB to the CLI: 19 MB/s before, 39 after (49 by the node's count);
uploads from the CLI 25 -> 29 MB/s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

4 files changed, 241 insertions, 0 deletions

diff --git a/packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py b/packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py
new file mode 100644
index 0000000..62835d3
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/client/sctp_fix.py
@@ -0,0 +1,126 @@
+"""Fixes to aiortc's SCTP stack, applied on import (aiortc 1.15).
+
+The first two are the node's (`meshbay_node/transport/webrtc/sctp_fix.py`, where
+why each is needed is written down); this is a copy because the client does not
+depend on the node, and `meshbay-common` does not depend on aiortc (the hub uses
+it without). `test_sctp_fix.py` in each package holds its copy to aiortc itself.
+
+In a download this side is the receiver of the node's 1 MiB `file_chunk`
+messages, about 870 SCTP packets each, and two receiver costs set the ceiling.
+Measured on loopback, 400 MB, one connection:
+
+- aiortc walked every fragment of the message in progress after each one
+  arrived, 153 million steps: two thirds of the client's CPU. Fixed by the
+  reassembly fix below — 20 MB/s became 28;
+- it acknowledged every packet. RFC 4960 §6.2 asks for one SACK every second
+  packet (Chromium does that), and each SACK costs the node a DTLS decryption
+  and a parse on the one core that also sends: more of its time than sending
+  the data. One every second packet — 28 MB/s became 39, the node's own figure
+  for the transfer 49.
+
+A SACK still goes at once when something is out of order or duplicated (that is
+the loss signal the sender is waiting for) or the packet carried anything other
+than data; a lone packet is acknowledged within `SACK_DELAY_S`.
+"""
+
+import asyncio
+import contextlib
+
+from aiortc.rtcsctptransport import (
+    SCTP_DATA_LAST_FRAG,
+    DataChunk,
+    InboundStream,
+    InitChunk,
+    RTCSctpTransport,
+    parse_packet,
+)
+
+# RFC 4960 §6.2: at most 500 ms; 200 is what implementations use.
+SACK_DELAY_S = 0.2
+
+
+def _flight_size_increase(self, chunk) -> None:
+    chunk._acked = False
+    self._flight_size += chunk._book_size
+
+
+RTCSctpTransport._flight_size_increase = _flight_size_increase
+
+
+_add_chunk = InboundStream.add_chunk
+_pop_messages = InboundStream.pop_messages
+_prune_chunks = InboundStream.prune_chunks
+
+
+def add_chunk(self, chunk) -> None:
+    _add_chunk(self, chunk)
+    if chunk.flags & SCTP_DATA_LAST_FRAG:
+        self._last_frags = getattr(self, "_last_frags", 0) + 1
+
+
+def pop_messages(self):
+    # Only ever too high (a pruned last fragment), never too low: a message
+    # leaves the reassembly with exactly one last fragment, and only here.
+    if not getattr(self, "_last_frags", 0):
+        return
+    for message in _pop_messages(self):
+        self._last_frags -= 1
+        yield message
+
+
+def prune_chunks(self, tsn: int) -> int:
+    size = _prune_chunks(self, tsn)
+    self._last_frags = sum(1 for c in self.reassembly if c.flags & SCTP_DATA_LAST_FRAG)
+    return size
+
+
+InboundStream.add_chunk = add_chunk
+InboundStream.pop_messages = pop_messages
+InboundStream.prune_chunks = prune_chunks
+
+
+async def _handle_data(self, data: bytes) -> None:
+    """aiortc's own, down to the SACK at the end: that one may wait for a
+    second packet."""
+    try:
+        _, _, verification_tag, chunks = parse_packet(data)
+    except ValueError:
+        return
+    if any(isinstance(x, InitChunk) for x in chunks):
+        assert len(chunks) == 1
+        expected_tag = 0
+    else:
+        expected_tag = self._local_verification_tag
+    if verification_tag != expected_tag:
+        return
+    for chunk in chunks:
+        await self._receive_chunk(chunk)
+    if not self._sack_needed:
+        return
+    self._sack_packets = getattr(self, "_sack_packets", 0) + 1
+    if (self._sack_packets >= 2 or self._sack_misordered or self._sack_duplicates
+            or not all(isinstance(c, DataChunk) for c in chunks)):
+        await _sack_now(self)
+    elif getattr(self, "_sack_timer", None) is None:
+        self._sack_timer = self._loop.call_later(SACK_DELAY_S, _sack_later, self)
+
+
+async def _sack_now(self) -> None:
+    timer = getattr(self, "_sack_timer", None)
+    if timer is not None:
+        timer.cancel()
+        self._sack_timer = None
+    self._sack_packets = 0
+    if self._sack_needed and self._association_state != self.State.CLOSED:
+        # The DTLS transport may be gone under a timer that fired late.
+        with contextlib.suppress(ConnectionError):
+            await self._send_sack()
+
+
+def _sack_later(self) -> None:
+    self._sack_timer = None
+    # Held: a task nobody references can be collected before it runs.
+    self._sack_task = asyncio.ensure_future(_sack_now(self))
+
+
+RTCSctpTransport._handle_data = _handle_data
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/session.py b/packages/meshbay-cli/src/meshbay_cli/client/session.py
index d10acc8..b6fc9e2 100644
--- a/packages/meshbay-cli/src/meshbay_cli/client/session.py
+++ b/packages/meshbay-cli/src/meshbay_cli/client/session.py
@@ -63,6 +63,7 @@ from meshbay_common.handshake import (
 from meshbay_common.join import join_transcript
 
 from ..errors import NODE, NODE_KEY, VERSION, CliError
+from . import sctp_fix, udp_buffer  # noqa: F401  (patch aiortc and aioice on import)
 from .account import Account
 from .keys import (
     LEGACY_MAGIC,
@@ -250,6 +251,13 @@ class GroupSession:
             raise CliError(NODE, "The group's index does not open under the group key; "
                                  "the session is ended.") from e
 
+    def open_delta(self, msg: dict) -> dict:
+        """A pushed `index_delta`, opened under the group key."""
+        try:
+            return unseal(self.gek, PURPOSE_INDEX, "index_delta", self.group_id, msg)
+        except Exception as e:
+            raise CliError(NODE, "An index update does not open under the group key.") from e
+
     async def close(self) -> None:
         with contextlib.suppress(Exception):
             await self.pc.close()
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py b/packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py
new file mode 100644
index 0000000..c0045d8
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/client/udp_buffer.py
@@ -0,0 +1,18 @@
+"""A larger receive buffer on the client's ICE sockets, applied on import.
+
+In a download the client is the receiver, and it reads its socket from Python
+— on the same event loop as the terminal interface — more slowly than a node
+sends in bursts. aioice asks the kernel for 256 KB per socket; what does not fit
+is dropped by this machine's kernel (`UdpRcvbufErrors`), the node's SCTP sender
+takes each drop for congestion and halves its window. Measured on loopback, 400
+MB: 3245 drops and 19 MB/s with 256 KB. The node does the same for the same
+reason in the other direction (`meshbay_node/transport/webrtc/udp_buffer.py`).
+
+Linux grants no more than `net.core.rmem_max`; Windows grants what is asked.
+"""
+
+import aioice.turn
+
+WANTED = 4 * 1024 * 1024
+
+aioice.turn.UDP_SOCKET_BUFFER_SIZE = WANTED
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]