1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
"""
`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 meshbay_common.crypto import generate_gek
from meshbay_node.indexer.group_index import GroupIndex
from conftest import one_root
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")
|