fix(node): make max_concurrent_streams take effect without a restart

`ops.set_node_settings` hot-swapped the stream pool by assigning
`webrtc._stream_sem`. That attribute has never existed on WebRTCTransport — the
pool is `ctx["_transcode_sem"]` — so `hasattr(webrtc, '_stream_sem')` was always
False and the branch never ran. The setting was accepted, written to roster.db
and node.toml, and applied only on the next restart, which is exactly what
draft-v6 §2.11 says it does not need. An operator lowering the cap on a
struggling machine, or raising it after "Server busy", saw nothing happen and
had no way to find out why.

`WebRTCTransport.set_capacity()` is the one implementation, on the object that
owns the state, so the download and upload caps the transfer-slots plan adds
next do not each grow their own copy of the mistake.

Resizing has semantics worth stating: the new cap governs new streams and never
interrupts one that is running, because a slot is held for the length of a film
and lowering a number must not take somebody's film away. The replacement pool
is built with the permits that remain (`new - in_flight`, floored at zero) — a
full set would briefly allow more concurrent viewers than either the old cap or
the new one.

That needs a count of slots in use, so `_stream_video` now maintains one instead
of the code reading the semaphore's private `_value`: a number this code keeps
itself survives the semaphore object being replaced underneath it, and the same
counter makes the "N of M in use" log lines mean something.

test_stream_capacity.py drives the real transport and the real `_stream_video`;
`test_ops_calls_the_real_mechanism` fails if the dead attribute comes back.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 155 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py
new file mode 100644
index 0000000..a35ece8
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_capacity.py
@@ -0,0 +1,155 @@
+"""
+`max_concurrent_streams` must take effect without a restart.
+
+`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an
+attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so
+`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the
+setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the
+Node page offers it as a live setting, and it did nothing: an operator lowering
+the cap on a struggling machine, or raising it after "Server busy", saw no
+change and had no way to know why.
+
+Nothing here mocks the pool. `set_capacity` is called on a real
+`WebRTCTransport` and the assertions read what a stream request would actually
+find.
+"""
+
+import asyncio
+
+import pytest
+
+from meshbay_node.transport.webrtc_server import (
+    MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, WebRTCTransport,
+)
+
+
+def _pool(transport) -> asyncio.Semaphore:
+    """The pool a stream request would acquire, built the way one builds it."""
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    return session._transcode_semaphore()
+
+
+@pytest.fixture
+def transport(tmp_path):
+    """A real WebRTCTransport. Its keys and index are genuine but incidental —
+    nothing below the capacity code reads them."""
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    from conftest import one_root
+    from meshbay_common.crypto import generate_gek
+    from meshbay_node.indexer.group_index import GroupIndex
+
+    sk_node = Ed25519PrivateKey.generate()
+    gek = generate_gek()
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    return WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=b"", gek=gek,
+        roots=one_root(shared),
+        index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek),
+        stun_servers=[])
+
+
+def test_raising_the_cap_is_visible_to_the_next_stream(transport):
+    """The bug, at its simplest: the number changes and nothing happens."""
+    pool = _pool(transport)
+    assert pool._value == MAX_CONCURRENT_TRANSCODES
+    transport.set_capacity(max_concurrent_streams=16)
+    assert _pool(transport)._value == 16, (
+        "the setting was accepted and the pool never changed — this is the "
+        "no-op that shipped")
+
+
+def test_lowering_the_cap_does_not_interrupt_what_is_running(transport):
+    """
+    A slot is held for the length of a film, so lowering the cap cannot take a
+    viewer's film away. It stops the next one starting, and the replacement pool
+    carries only the permits that remain.
+    """
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 3
+    transport.set_capacity(max_concurrent_streams=4)
+    assert _pool(transport)._value == 1, (
+        "a full set of permits would let more viewers in than either the old "
+        "cap or the new one, on top of the three still watching")
+
+
+def test_lowering_below_what_is_running_refuses_the_next_one(transport):
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 6
+    transport.set_capacity(max_concurrent_streams=2)
+    assert _pool(transport)._value == 0, "the pool must not go negative"
+
+
+def test_the_value_is_kept_for_a_pool_not_yet_built(transport):
+    """Nothing has streamed, so there is nothing to resize — but the number has
+    to be there when the first request builds the pool."""
+    transport.set_capacity(max_concurrent_streams=3)
+    assert transport._ctx.get("_transcode_sem") is None
+    assert _pool(transport)._value == 3
+
+
+def test_a_cap_below_one_is_refused(transport):
+    for bad in (0, -1):
+        with pytest.raises(ValueError):
+            transport.set_capacity(max_concurrent_streams=bad)
+
+
+def test_nothing_changes_when_nothing_is_passed(transport):
+    _pool(transport)
+    before = transport._ctx["_transcode_sem"]
+    assert transport.set_capacity() == {}
+    assert transport._ctx["_transcode_sem"] is before
+
+
+@pytest.mark.asyncio
+async def test_in_flight_is_counted_by_the_streaming_path_itself(transport):
+    """
+    `set_capacity` resizes against `_streams_in_flight`, so that counter has to
+    be maintained where slots are actually taken — not set by a test. Drives the
+    real `_stream_video`, with the work under it stubbed: what is being checked
+    is the accounting around the slot, which is where flow control in this repo
+    has gone wrong before.
+    """
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    session._send = lambda msg: None
+
+    seen = []
+    release = asyncio.Event()
+
+    async def _inner(_msg):
+        seen.append(transport._ctx.get("_streams_in_flight"))
+        await release.wait()
+
+    session._stream_video_inner = _inner
+    task = asyncio.create_task(session._stream_video({"file_id": "x"}))
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert seen == [1], "the slot was taken without being counted"
+
+    release.set()
+    await task
+    assert transport._ctx["_streams_in_flight"] == 0, (
+        "a slot that is not given back is a viewer nobody can replace — the "
+        "class of bug _replace_stream and shutdown_tasks exist for")
+
+
+def test_ops_calls_the_real_mechanism():
+    """
+    The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every
+    WebRTCTransport that has ever existed, so a test that only checked
+    "set_node_settings does not raise" passed throughout.
+    """
+    import inspect
+
+    from meshbay_node import ops
+
+    src = inspect.getsource(ops.set_node_settings)
+    # Comments stripped: this function now *explains* the dead attribute, and a
+    # test that matched the prose would fail on its own documentation.
+    code = "\n".join(line.split("#", 1)[0] for line in src.splitlines())
+    assert "_stream_sem" not in code, "the attribute that never existed is back"
+    assert "set_capacity" in code, "the setting must reach the pool that exists"
+    assert not hasattr(WebRTCTransport, "_stream_sem")