From 9d38418817d1894b31c93ece0113e503d82167cd Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 11 Oct 2026 15:42:52 +0200 Subject: feat(node): log what a transfer moved when it ends One INFO line per lease: bytes, duration, rate, and the SCTP association's window, round trip, losses and T3 expirations. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-node/tests/test_lease_enforcement.py | 33 ++++++++++ packages/meshbay-node/tests/test_sctp_stats.py | 74 ++++++++++++++++++++++ packages/meshbay-node/tests/test_transfer_slots.py | 23 +++++++ 3 files changed, 130 insertions(+) create mode 100644 packages/meshbay-node/tests/test_sctp_stats.py (limited to 'packages/meshbay-node/tests') diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py index 2b1014b..7b636ac 100644 --- a/packages/meshbay-node/tests/test_lease_enforcement.py +++ b/packages/meshbay-node/tests/test_lease_enforcement.py @@ -273,3 +273,36 @@ async def test_the_sweeper_does_not_reclaim_a_transfer_that_is_moving(group): await _request(peer, _ids(group)[0], tr="t1") assert slots.leases["t1"].used is True await asyncio.sleep(0) + + +# ── the line the node logs when a transfer ends ───────────────────────────── + +async def test_a_served_chunk_is_counted_on_its_lease(group): + peer = _Session(group) + group["_peers"]["s1"] = peer + peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + await _request(peer, _ids(group)[0], tr="t1") + await _request(peer, _ids(group)[0], tr="invented") + + lease = peer._slots().leases["t1"] + assert lease.moved == len(peer.chunks()[0]["ct"]) + line = peer._transfer_end_line(lease, "done") + assert line.startswith("transfer: end download t1 done 1.0 MB"), line + + +async def test_a_closed_transfer_is_logged_once(group): + peer = _Session(group) + group["_peers"]["s1"] = peer + lines: list[str] = [] + end_line = peer._transfer_end_line + peer._transfer_end_line = ( + lambda lease, reason: lines.append(end_line(lease, reason)) or lines[-1]) + + peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + peer._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000}) + peer._do_transfer_close({"tr": "t1", "reason": "cancelled"}) + peer._release_transfers() + peer._release_transfers() + + assert [x.split()[3:5] for x in lines] == [ + ["t1", "cancelled"], ["t2", "session_gone"]], lines diff --git a/packages/meshbay-node/tests/test_sctp_stats.py b/packages/meshbay-node/tests/test_sctp_stats.py new file mode 100644 index 0000000..bf8970d --- /dev/null +++ b/packages/meshbay-node/tests/test_sctp_stats.py @@ -0,0 +1,74 @@ +"""Losses and T3 expirations are counted per association (see sctp_stats.py). + +Replayed against aiortc's own RTCSctpTransport: five chunks sent and the first +one lost, so three acknowledgements in a row report it missing and the sender +enters fast recovery, once. Then T3 expires, once. +""" + +import asyncio + +from aiortc.rtcsctptransport import RTCSctpTransport, SackChunk +from meshbay_node.transport.webrtc import sctp_stats + + +class _Dtls: + state = "connected" + + def _register_data_receiver(self, *a): + pass + + +class _Pc: + def __init__(self, sctp): + self.sctp = sctp + + +def _sack(cumulative_tsn: int, gaps: list) -> SackChunk: + sack = SackChunk() + sack.cumulative_tsn = cumulative_tsn % 2**32 + sack.gaps = gaps + sack.duplicates = [] + sack.advertised_rwnd = 1 << 20 + return sack + + +async def _replay() -> RTCSctpTransport: + s = RTCSctpTransport(_Dtls()) + s._association_state = s.State.ESTABLISHED + + async def sent(*_a, **_k): + pass + + s._send_chunk = sent + for name in ("_t3_start", "_t3_restart", "_t3_cancel"): + setattr(s, name, lambda: None) + s._cwnd = 100_000 + + for _ in range(5): + await s._send(1, 51, b"x" * 1000) + first = s._sent_queue[0].tsn + for last_seen in (2, 3, 4): + await s._receive_sack_chunk(_sack(first - 1, [(2, last_seen + 1)])) + # Still recovering from the same loss: one episode, not a second. + await s._receive_sack_chunk(_sack(first - 1, [(2, 5)])) + s._t3_expired() + return s + + +def test_a_loss_and_a_t3_expiry_are_each_counted_once(): + s = asyncio.run(_replay()) + assert s._mb_losses == 1 + assert s._mb_t3 == 1 + assert sctp_stats.describe(_Pc(s)).endswith("loss=1 t3=1") + + +def test_an_association_that_lost_nothing_says_so(): + async def fresh() -> RTCSctpTransport: + return RTCSctpTransport(_Dtls()) + + s = asyncio.run(fresh()) + assert sctp_stats.describe(_Pc(s)).endswith("rtt=? loss=0 t3=0") + + +def test_a_session_without_an_association_adds_nothing(): + assert sctp_stats.describe(None) == "" diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py index 08e7e70..68d8858 100644 --- a/packages/meshbay-node/tests/test_transfer_slots.py +++ b/packages/meshbay-node/tests/test_transfer_slots.py @@ -369,3 +369,26 @@ def test_using_a_lease_forgives_its_earlier_misses(_=None): assert s.leases["slow"].missed_grants == 2 s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3) assert s.leases["slow"].missed_grants == 0 + + +# ── what the end-of-transfer line is built from ───────────────────────────── + +def test_a_lease_counts_what_moved_and_for_how_long(): + slots = TransferSlots() + slots.open(tr="t1", kind=DOWNLOAD, session_key="s1", user_id="alice", + group_id="g1", now=0.0) + slots.touch("t1", now=10.0) + slots.account("t1", 1_000_000, now=10.5) + slots.touch("t1", now=11.0) + slots.account("t1", 500_000, now=12.0) + lease = slots.leases["t1"] + assert lease.moved == 1_500_000 + assert lease.elapsed == 2.0 + + +def test_a_lease_nothing_moved_under_has_no_duration(): + slots = TransferSlots() + lease, _ = slots.open(tr="t1", kind=DOWNLOAD, session_key="s1", + user_id="alice", group_id="g1", now=0.0) + assert (lease.moved, lease.elapsed) == (0, 0.0) + slots.account("unknown", 10) -- cgit v1.2.3