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
157
158
159
160
161
|
"""
Switching video before the first one finishes.
Reported from a phone: play a video, do not wait for the end, open another —
the player sits on "buffering" and never recovers. Two independent causes, both
in how one stream hands over to the next.
**The flag that outlived its stream.** `flushQueue` returns early while an
append is in flight (`appendingRef`). Teardown reset the queue, the SourceBuffer
and the MediaSource, but not that flag. Switch while an append was running and
it stayed true: the new SourceBuffer never received anything, so no `updateend`
ever cleared it, no credit ever went back to the node, the node stopped sending
and the phase never left "buffering". More likely on a phone, where an append
takes long enough to still be running when a finger moves.
`endedRef` had the same shape: left true, the next stream calls `endOfStream()`
the first time its queue runs dry and truncates the film.
**Segments from the film you left.** The DataChannel is ordered, so whatever the
node had already sent arrives after the switch. Every stream message carries a
`file_id` and nothing looked at it, so those segments were decrypted against the
new file — failing, in the console, for data that was simply not ours.
The first half is checked by running the real `flushQueue` guard against the
real reset, in node. The second is read from the source, since it is a shape
rather than a behaviour.
"""
import json
import re
import shutil
import subprocess
from pathlib import Path
import pytest
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
APP = STATIC / "video-player.js"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not APP.exists(),
reason="node or the SPA sources are not available")
@pytest.fixture(scope="module")
def app():
return APP.read_text()
def _player(app: str) -> str:
i = app.index("function VideoPlayer(")
nxt = app.find("\nfunction ", i + 1)
return app[i:nxt if nxt > 0 else len(app)]
# ── The flag that stalled everything ──────────────────────────────────────────
def test_a_new_stream_starts_from_a_clean_slate(app):
"""The reset must be where the stream starts, not where the last one ended.
Teardown is skippable — an unmount that races, an effect that re-runs — and
a stale flag costs the whole player. Starting from a known state cannot be
skipped.
"""
player = _player(app)
effect = player[player.index("useEffect(() => {\n let cancelled = false;"):]
effect = effect[:effect.index("transport.requestStream")]
for ref in ("appendingRef", "endedRef"):
assert f"{ref}.current = false" in effect, (
f"{ref} is not reset before the stream starts")
def test_the_stall_is_reproduced_and_the_reset_clears_it(tmp_path):
"""The guard and the reset, run for real rather than read.
Models flushQueue's first line and the append/updateend cycle. Without the
reset the second stream appends nothing at all; with it, it drains.
"""
script = tmp_path / "case.mjs"
script.write_text("""
// flushQueue's guard, and the part that makes it a trap: `updateend`
// belongs to the SourceBuffer, which teardown throws away, while
// `appending` is a ref on the component, which survives.
const makePlayer = (resetOnStart) => {
const st = { appending: false, appended: 0, queue: [], owed: 0 };
return {
st,
startStream() { if (resetOnStart) st.appending = false; },
push(seg) { st.queue.push(seg); this.flush(); },
flush() {
if (st.appending || st.queue.length === 0) return;
st.appending = true;
st.queue.shift();
st.appended++;
st.owed++; // the browser will fire updateend for this
},
fireUpdateend() { // only for appends that really happened
while (st.owed > 0) {
st.owed--;
st.appending = false;
this.flush();
}
},
teardown() {
st.queue = [];
st.owed = 0; // the SourceBuffer and its listener are gone
},
};
};
const out = {};
for (const [name, reset] of [["without_reset", false], ["with_reset", true]]) {
const p = makePlayer(reset);
// First video: an append is in flight and its updateend has not arrived
// when the viewer moves on. That is the whole scenario.
p.startStream();
p.push("A1");
p.teardown();
const afterFirst = p.st.appended;
// Second video.
p.startStream();
p.push("B1"); p.fireUpdateend();
p.push("B2"); p.fireUpdateend();
out[name] = p.st.appended - afterFirst;
}
console.log(JSON.stringify(out));
""")
proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
assert proc.returncode == 0, proc.stderr
got = json.loads(proc.stdout)
assert got["without_reset"] == 0, (
"the scenario no longer reproduces the stall, so this test proves nothing")
assert got["with_reset"] == 2, (
"the second video still cannot append — the reset does not clear the stall")
# ── Segments from the abandoned stream ────────────────────────────────────────
@pytest.mark.parametrize("handler", ["onStreamInit", "onStreamData", "onStreamEnd"])
def test_stream_messages_are_matched_to_the_file_they_belong_to(app, handler):
player = _player(app)
body = player[player.index(f"transport.{handler} = "):]
body = body[:body.index("\n };")]
assert "file_id !== entry.id" in body, (
f"{handler} accepts a message from any stream — on an ordered channel "
"the film you just left is still arriving")
def test_the_node_actually_stamps_those_messages():
"""The guard above is worth nothing if the field is not sent."""
server = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src"
/ "meshbay_node" / "transport" / "webrtc_server.py")
if not server.exists():
pytest.skip("the node sources are not available")
text = server.read_text()
for const in ("STREAM_INIT", "STREAM_DATA", "STREAM_END"):
i = text.index(f"MNP.{const}")
block = text[i:i + 400]
assert re.search(r'"file_id":\s*file_id', block), (
f"{const} carries no file_id, so the client cannot tell streams apart")
|