Merge branch 'main' of meshbay.org:meshbay

4 files changed, 319 insertions, 28 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index c5596f1..d46ab8c 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -74,13 +74,14 @@ max_concurrent_uploads = 8
 # the partial file is deleted.
 max_upload_gb = 8
 
-# How many processes carry the connections of members. At 1 the node does it
+# How many processes carry the connections of members. Left unset: one per core
+# of this machine, up to four, about 85 MB of memory each. At 1 the node does it
 # all in one process, on one core: about 35 MB/s summed over every transfer in
-# progress. Above 1 the node starts that many processes for the connections and
-# keeps the rest for itself, so transfers use that many cores, plus part of one
-# for the node. Takes effect for new connections, without a restart. Not above
-# the number of cores of this machine.
-connection_processes = 1
+# progress on Linux, and much less on Windows. Above 1 the node starts that many
+# processes for the connections and keeps the rest for itself, so transfers use
+# that many cores, plus part of one for the node. Takes effect for new
+# connections, without a restart. Not above the number of cores of this machine.
+# connection_processes = 4
 
 # HEVC sources have no browser decoder on most platforms, so streaming one is
 # transcoded to H264 rather than the usual free copy — real CPU per viewer.
@@ -202,8 +203,9 @@ class NodeConfig:
     # §15.3), so this is what stands between a writable root and a full disk.
     max_upload_gb: float = 8.0
     # How many processes hold the peer connections. One means this process
-    # holds them itself, as it always did. See transport/webrtc/workers.py.
-    connection_processes: int = 1
+    # holds them itself. See transport/webrtc/workers.py, and
+    # `default_connection_processes` for why it is not one by default.
+    connection_processes: int = field(default_factory=lambda: default_connection_processes())
     # HEVC (and any future codec in media_probe.py's
     # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so
     # streaming it needs a real re-encode to H264 rather than the usual free
@@ -305,6 +307,23 @@ def max_processes() -> int:
     return max(1, os.cpu_count() or 1)
 
 
+# Above four, a single download no longer gains (docs/MESHBAY_DESIGN.md §5.5:
+# 131 MB/s at four, 150 at eight with the receiving browser the limit), and each
+# process is 85 MB held for the life of the node.
+DEFAULT_CONNECTION_PROCESSES_MAX = 4
+
+
+def default_connection_processes() -> int:
+    """
+    One per core, up to four. It was 1, which starts no process at all: the
+    several-cores path was measured, shipped and reached by no node whose
+    operator had not found the setting, and the lanes of a download all landed
+    on the node's one core — on Windows, where Python's UDP costs most, a
+    download over eight connections was as slow as over one.
+    """
+    return min(DEFAULT_CONNECTION_PROCESSES_MAX, max_processes())
+
+
 def _positive(value: object, default: int, name: str) -> int:
     """A count that must be at least one, or the default with a word about it.
 
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py
index 0d45a7d..a7e6f86 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_fix.py
@@ -1,12 +1,13 @@
-"""A fix to aiortc's SCTP sender, applied on import (aiortc 1.15).
+"""Two fixes to aiortc's SCTP stack, applied on import (aiortc 1.15).
 
-aiortc counts the bytes in flight and sends nothing new while that count is at
-or above the congestion window. A chunk the peer acknowledged in a gap block is
-marked `_acked` and taken out of the count. If the T3 timer then expires, every
-chunk still queued is sent again, that one included, and goes back into the
-count with `_acked` still set. When the cumulative ack finally covers it, the
-SACK handler skips it because it reads as acknowledged already, and its bytes
-stay counted for the life of the association.
+**The sender leaked bytes in flight.** aiortc counts the bytes in flight and
+sends nothing new while that count is at or above the congestion window. A chunk
+the peer acknowledged in a gap block is marked `_acked` and taken out of the
+count. If the T3 timer then expires, every chunk still queued is sent again,
+that one included, and goes back into the count with `_acked` still set. When
+the cumulative ack finally covers it, the SACK handler skips it because it reads
+as acknowledged already, and its bytes stay counted for the life of the
+association.
 
 Each such loss leaks a chunk. On a lossy path (a phone on Wi-Fi) the leak grows
 past the window, which T3 itself keeps small. From then on the node sends
@@ -14,14 +15,49 @@ nothing on that connection: nothing is outstanding, so no timer runs to unstick
 it, and the association still reads as established. The peer keeps talking and
 the node keeps answering into a queue that never moves. Live symptom: the photo
 backup stalled after a few dozen photos, its slot granted and the grant never
-delivered (docs/MESHBAY_DESIGN.md §9.12).
+delivered (docs/MESHBAY_DESIGN.md §9.12). The fix is the invariant the code
+assumed: a chunk being sent again is in flight, so it is not acknowledged.
 
-The fix is the invariant the code assumed: a chunk being sent again is in
-flight, so it is not acknowledged. `test_sctp_fix.py` replays the sequence
-against aiortc itself.
+**The receiver reassembled in quadratic time.** After every DATA chunk,
+`InboundStream.pop_messages` walks the reassembly list from its start, through
+every fragment of the message still being assembled. A message of n fragments
+costs n²/2 steps, and a 1 MiB `file_chunk` is about 870 fragments: measured on a
+receiving aiortc, 153 million steps for 400 MB, two thirds of its CPU, and the
+whole of why it could not take more than 20 MB/s on loopback. No message can
+come out of a reassembly holding no last fragment, so the walk is skipped until
+one is there; when one is, the walk is aiortc's own, unchanged.
+
+The node receives uploads in 48 KB messages, where the walk is short; the
+command-line client receives 1 MiB ones and carries the same fix
+(`meshbay_cli/client/sctp_fix.py`).
+
+**The receiver acknowledged every packet.** RFC 4960 §6.2 asks for one SACK
+every second packet, and Chromium does that; aiortc sent one at the end of
+every packet of data. Each SACK costs its receiver a DTLS decryption and a parse
+on the one core that also does everything else — on a node sending a download
+to the command-line client, more time than sending the data. In an upload the
+node is the receiver and pays the other half: the encryption and the send of a
+SACK per packet. A SACK still goes at once when something is out of order or
+duplicated (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`.
+
+`test_sctp_fix.py` replays all three against aiortc itself.
 """
 
-from aiortc.rtcsctptransport import RTCSctpTransport
+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:
@@ -30,3 +66,82 @@ def _flight_size_increase(self, chunk) -> None:
 
 
 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-node/tests/test_connection_processes.py b/packages/meshbay-node/tests/test_connection_processes.py
index ab9dfd3..778da5a 100644
--- a/packages/meshbay-node/tests/test_connection_processes.py
+++ b/packages/meshbay-node/tests/test_connection_processes.py
@@ -13,6 +13,7 @@ not carry.
 """
 
 import asyncio
+import re
 import struct
 from pathlib import Path
 
@@ -231,11 +232,21 @@ async def test_one_process_starts_none():
 
 # ── the setting ─────────────────────────────────────────────────────────────
 
-def test_the_default_is_one_process(tmp_path):
-    assert node_settings_defaults()["connection_processes"] == 1
+@pytest.mark.parametrize("cores, expected", [(1, 1), (2, 2), (4, 4), (28, 4)])
+def test_the_default_is_one_per_core_up_to_four(tmp_path, monkeypatch, cores, expected):
+    """It was 1, which starts none: the path was shipped and reached by no node
+    whose operator had not found the setting."""
+    monkeypatch.setattr("meshbay_node.config.max_processes", lambda: cores)
+    assert node_settings_defaults()["connection_processes"] == expected
     conf = tmp_path / "node.toml"
     conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n')
-    assert load_config(conf).node.connection_processes == 1
+    assert load_config(conf).node.connection_processes == expected
+
+
+def test_a_new_node_toml_leaves_it_to_the_default(tmp_path):
+    from meshbay_node.config import EXAMPLE_CONFIG
+    assert not re.search(r"^connection_processes\s*=", EXAMPLE_CONFIG, re.M), (
+        "a value written by `init` is a value no later default reaches")
 
 
 def test_node_toml_cannot_ask_for_more_than_the_cores(tmp_path):
@@ -296,4 +307,5 @@ async def test_a_number_the_machine_cannot_honour_is_refused(state, monkeypatch,
     with pytest.raises(ops.OpError):
         await ops.set_node_settings(state, {"connection_processes": value})
     assert state["webrtc"].asked == []
-    assert (await ops.get_node_settings(state))["connection_processes"] == 1
+    assert ((await ops.get_node_settings(state))["connection_processes"]
+            == state["config"].node.connection_processes)
diff --git a/packages/meshbay-node/tests/test_sctp_fix.py b/packages/meshbay-node/tests/test_sctp_fix.py
index 0b2536a..c3fe622 100644
--- a/packages/meshbay-node/tests/test_sctp_fix.py
+++ b/packages/meshbay-node/tests/test_sctp_fix.py
@@ -1,4 +1,5 @@
-"""aiortc's SCTP sender must not leak bytes in flight (see sctp_fix.py).
+"""aiortc's SCTP stack as sctp_fix.py leaves it: no bytes leaked in flight,
+reassembly in linear time, one SACK every second packet.
 
 Replayed against aiortc's own RTCSctpTransport: three chunks sent, the first
 lost and the other two acknowledged in a gap block, T3 expires and they are
@@ -8,10 +9,20 @@ enough of those stop the node from sending anything on that connection.
 """
 
 import asyncio
-
-from aiortc.rtcsctptransport import RTCSctpTransport, SackChunk
+import random
 
 import meshbay_node.transport.webrtc.core  # noqa: F401  (applies the fix)
+from aiortc.rtcsctptransport import (
+    SCTP_DATA_FIRST_FRAG,
+    SCTP_DATA_LAST_FRAG,
+    SCTP_DATA_UNORDERED,
+    DataChunk,
+    InboundStream,
+    RTCSctpTransport,
+    SackChunk,
+    serialize_packet,
+)
+from meshbay_node.transport.webrtc import sctp_fix
 
 
 class _Dtls:
@@ -57,3 +68,137 @@ def test_nothing_outstanding_means_nothing_in_flight():
     s = asyncio.run(_replay())
     assert not s._sent_queue
     assert s._flight_size == 0
+
+
+# ── Reassembly (see sctp_fix.py) ─────────────────────────────────────────────
+
+
+def _fragments(messages: list[int], *, unordered: bool = False, tsn: int = 1000):
+    """DATA chunks for messages of the given fragment counts, in TSN order."""
+    out = []
+    for seq, n in enumerate(messages):
+        for i in range(n):
+            c = DataChunk()
+            c.flags = ((SCTP_DATA_FIRST_FRAG if i == 0 else 0)
+                       | (SCTP_DATA_LAST_FRAG if i == n - 1 else 0)
+                       | (SCTP_DATA_UNORDERED if unordered else 0))
+            c.tsn, c.stream_id, c.stream_seq, c.protocol = tsn, 1, seq, 53
+            c.user_data = f"{seq}.{i};".encode()
+            tsn += 1
+            out.append(c)
+    return out
+
+
+class _Original:
+    """aiortc's own reassembly, untouched, for the same arrivals."""
+
+    def __init__(self):
+        self.reassembly, self.sequence_number = [], 0
+
+    def add(self, chunk):
+        sctp_fix._add_chunk(self, chunk)
+        return list(sctp_fix._pop_messages(self))
+
+
+def test_reassembly_gives_what_aiortc_gives_for_any_arrival_order():
+    rng = random.Random(7)
+    for trial in range(300):
+        messages = [rng.randint(1, 12) for _ in range(rng.randint(1, 8))]
+        chunks = _fragments(messages, unordered=trial % 3 == 0)
+        # Reordered within a short window, as a network reorders.
+        for i in range(len(chunks)):
+            j = min(len(chunks) - 1, i + rng.randint(0, 4))
+            chunks[i], chunks[j] = chunks[j], chunks[i]
+        fixed, original = InboundStream(), _Original()
+        got, want = [], []
+        for c in chunks:
+            fixed.add_chunk(c)
+            got.append(list(fixed.pop_messages()))
+            want.append(original.add(c))
+        assert got == want, (trial, messages)
+
+
+def test_a_large_message_is_walked_once_not_once_per_fragment(monkeypatch):
+    walks = []
+    original = sctp_fix._pop_messages
+
+    def counted(stream):
+        walks.append(len(stream.reassembly))
+        return original(stream)
+
+    monkeypatch.setattr(sctp_fix, "_pop_messages", counted)
+    s = InboundStream()
+    out = []
+    for c in _fragments([1000, 1000]):
+        s.add_chunk(c)
+        out += list(s.pop_messages())
+    assert len(out) == 2
+    assert walks == [1000, 1000], "one walk per message, when its last fragment is in"
+
+
+def test_pruning_a_last_fragment_does_not_stall_the_next_message():
+    s = InboundStream()
+    first, second = _fragments([3]), _fragments([2], tsn=1003)
+    second[0].stream_seq = second[1].stream_seq = 1
+    for c in first[1:]:          # the first fragment never comes
+        s.add_chunk(c)
+        assert not list(s.pop_messages())
+    s.prune_chunks(first[-1].tsn)  # abandoned (FORWARD TSN)
+    s.sequence_number = 1
+    for c in second:
+        s.add_chunk(c)
+    assert [m[2] for m in s.pop_messages()] == [b"0.0;0.1;"]
+
+
+# ── Delayed acknowledgement (see sctp_fix.py) ────────────────────────────────
+
+def _receiver():
+    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
+        s._sack_duplicates.clear()
+
+    async def deliver(*_a):
+        pass
+
+    s._send_sack = send_sack
+    s._receive = deliver
+    return s, sacks
+
+
+def _packet(s, chunk) -> bytes:
+    return serialize_packet(5000, 5000, s._local_verification_tag, chunk)
+
+
+def test_one_sack_every_second_packet_of_data():
+    async def go():
+        s, sacks = _receiver()
+        for c in _fragments([1] * 10):
+            await s._handle_data(_packet(s, c))
+        return sacks
+    assert asyncio.run(go()) == [1001, 1003, 1005, 1007, 1009]
+
+
+def test_a_gap_is_acknowledged_at_once():
+    async def go():
+        s, sacks = _receiver()
+        chunks = _fragments([1] * 4)
+        await s._handle_data(_packet(s, chunks[0]))
+        await s._handle_data(_packet(s, chunks[2]))   # 1001 missing
+        return sacks
+    assert asyncio.run(go()) == [1000], "the gap is the loss signal the sender waits for"
+
+
+def test_a_lone_packet_is_acknowledged_within_the_delay():
+    async def go():
+        s, sacks = _receiver()
+        await s._handle_data(_packet(s, _fragments([1])[0]))
+        assert sacks == []
+        await asyncio.sleep(sctp_fix.SACK_DELAY_S + 0.1)
+        return sacks
+    assert asyncio.run(go()) == [1000]