playlists: drive the blob ops over a real DataChannel

The six MNP 3.1 ops were the only new ones never crossing a channel in a
test. Two cases on the existing aiortc harness: a round trip read back on
a second connection, and a 256 KB body. Found that the 1 MB body cap is
unreachable from a browser — docs/playlists.md §15.3.

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

1 files changed, 179 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index 284c461..3010fe9 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -9,6 +9,7 @@ Uses local loopback (no STUN/ICE needed for localhost).
 
 import asyncio
 import base64
+from contextlib import asynccontextmanager
 import hashlib
 import hmac
 import os
@@ -1770,3 +1771,181 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_
     await bundle_store.close()
     await browser_pc.close()
     await transport.close_all()
+
+
+# ── Per-account blobs, MNP 3.1 (docs/playlists.md §15.2) ─────────────────────
+#
+# `test_user_blob_mnp.py` drives these six handlers directly, which proves what
+# they decide but nothing about how their payloads travel: it never builds a
+# frame and never crosses a channel. A playlist body is the largest `bin` value
+# this protocol carries after a file chunk, and the failure it would hit is
+# silent — a playlist that does not come back reports no error, it is simply
+# absent, which is exactly what was seen on a phone once already.
+
+
+@asynccontextmanager
+async def _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """
+    A node with a bundle store, and peers, torn down whatever happens.
+
+    The `finally` is not tidiness. `BundleStore` runs an aiosqlite thread, and a
+    test that fails before closing it leaves that thread alive — the process
+    then hangs in `threading._shutdown`, *after* pytest has printed the failure
+    and the summary. Found by breaking the node on purpose to check these tests
+    catch it: they did, in 0.66s, and then the run never ended. A red suite is a
+    result; a stuck one is an outage.
+    """
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g",
+                               sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek,
+        roots=one_root(shared_dir), index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["bundle_store"] = bundle_store
+
+    peers: list = []
+    try:
+        yield transport, peers
+    finally:
+        for pc in peers:
+            await pc.close()
+        await bundle_store.close()
+        await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_user_blobs_round_trip_over_a_live_datachannel(
+        sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """
+    Sealed bytes out, the same bytes back, across two connections.
+
+    The blobs are `os.urandom`, deliberately: sealed output is incompressible
+    and uses every byte value, so anything that treats this as text — a UTF-8
+    decode, msgpack `str` instead of `bin` — corrupts it. Bytes the node could
+    have round-tripped by accident would prove nothing.
+
+    The second connection is the case this feature exists for: a device that was
+    not the one that wrote.
+    """
+    async with _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path) as (
+            transport, peers):
+        manifest = os.urandom(512)
+        # Bodies are padded to a multiple of 4 KB before sealing, so this is the
+        # smallest one a real client ever sends.
+        body = os.urandom(4096)
+
+        pc1, ch1, q1 = await _setup_peer(transport, sk_hub, gek, "peer-blob-write")
+        peers.append(pc1)
+
+        ch1.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION,
+                        "kind": "playlists", "rev": 7, "blob_enc": manifest}))
+        ack = await asyncio.wait_for(q1.get(), timeout=5.0)
+        assert ack["type"] == "ack"
+        assert ack["detail"] == "user_blob_stored"
+
+        ch1.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION,
+                        "kind": "playlist:favorites", "rev": 3, "blob_enc": body}))
+        ack = await asyncio.wait_for(q1.get(), timeout=5.0)
+        assert ack["detail"] == "user_blob_stored"
+
+        await pc1.close()
+
+        # A second device, same account, which has never seen either blob.
+        pc2, ch2, q2 = await _setup_peer(transport, sk_hub, gek, "peer-blob-read")
+        peers.append(pc2)
+
+        # What is here, and at what revision. This is the call that lets a fresh
+        # device discover kinds it cannot guess — they carry client-made ids —
+        # and it must not carry payloads.
+        ch2.send(_pack({"type": MNP.USER_BLOB_LIST, "v": MNP_VERSION}))
+        listing = await asyncio.wait_for(q2.get(), timeout=5.0)
+        assert listing["type"] == MNP.USER_BLOB_LIST_RESP
+        assert listing["blobs"] == [{"kind": "playlist:favorites", "rev": 3},
+                                    {"kind": "playlists", "rev": 7}]
+
+        # `req_id` is what lets the client match a reply to its request rather
+        # than to whatever arrives next; over a channel carrying several
+        # requests at once, arrival order is not an answer.
+        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
+                        "kind": "playlists", "req_id": "r-91"}))
+        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
+        assert resp["type"] == MNP.USER_BLOB_RESP
+        assert resp.get("req_id") == "r-91"
+        assert resp["kind"] == "playlists"
+        assert resp["rev"] == 7
+        assert isinstance(resp["blob_enc"], bytes), "came back as something else than bin"
+        assert resp["blob_enc"] == manifest
+
+        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
+                        "kind": "playlist:favorites"}))
+        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
+        assert resp["rev"] == 3
+        assert resp["blob_enc"] == body
+
+        # A kind never written is an absence, not an error: the ordinary state
+        # of a fresh node.
+        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
+                        "kind": "playlist:never-written"}))
+        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
+        assert resp["rev"] is None
+        assert resp["blob_enc"] is None
+
+        # And a delete reclaims it, which is how a tombstoned playlist stops
+        # costing an account its quota.
+        ch2.send(_pack({"type": MNP.USER_BLOB_DELETE, "v": MNP_VERSION,
+                        "kind": "playlist:favorites"}))
+        ack = await asyncio.wait_for(q2.get(), timeout=5.0)
+        assert ack["detail"] == "user_blob_deleted"
+
+        ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
+                        "kind": "playlist:favorites"}))
+        resp = await asyncio.wait_for(q2.get(), timeout=5.0)
+        assert resp["blob_enc"] is None
+
+
+@pytest.mark.asyncio
+async def test_a_large_user_blob_round_trips_whole(
+        sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """
+    A quarter-megabyte blob, out and back, byte for byte.
+
+    This is the risk §15.2 named, and measuring it corrected how it was
+    described: a 256 KB blob arrives as **one** application message of 262213
+    bytes, not as fragments the four-byte length prefix reassembles. SCTP
+    fragments it and puts it back together underneath. What this pins is that
+    nothing in the node's own framing truncates or re-encodes a `bin` value of
+    that size.
+
+    Note which direction is proved. A *browser* cannot send a frame this large
+    to this node: aiortc advertises `a=max-message-size:65536`, so Chrome
+    refuses anything above it — which is why uploads chunk at 48 KB
+    (`transport.js:41`). The store below is aiortc talking to aiortc and says
+    nothing about that ceiling. The fetch is the direction that matters here: a
+    node answers with a whole body, and a long playlist is the largest one.
+    """
+    async with _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path) as (
+            transport, peers):
+        # 256 KB: about nine hundred tracks at the 270 bytes a sealed track
+        # measures.
+        body = os.urandom(256 * 1024)
+
+        pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-blob-big")
+        peers.append(pc)
+
+        ch.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION,
+                       "kind": "playlist:long", "rev": 1, "blob_enc": body}))
+        ack = await asyncio.wait_for(q.get(), timeout=10.0)
+        assert ack["detail"] == "user_blob_stored", ack
+
+        ch.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION,
+                       "kind": "playlist:long"}))
+        resp = await asyncio.wait_for(q.get(), timeout=10.0)
+        assert resp["type"] == MNP.USER_BLOB_RESP
+        assert len(resp["blob_enc"]) == len(body), "truncated"
+        assert resp["blob_enc"] == body