foxyzandClaude Opus 5.5
ec0849eb01
Before, the member passing a page link on downloaded it twice, independently: their mpv straight from the CDN and the relay's pulls again, with a different format pick and without yt-dlp's cookies. The two drifted (the putter played on while viewers stalled, or the reverse) and doubled the connections the CDN saw. Now one SharedSource per track reads the site once (resuming after cuts, backing off on 412/429/5xx and silence) into a bounded buffer (64 MiB ahead, 16 behind) that serves both the putter's mpv through a loopback endpoint and the server's pulls; the putter plays the relay's own format pick, and the relay sends yt-dlp's cookies. Up next's warm relay uses the same reader. Pieces tracks (HLS/DASH) keep two fetches for now. Measured in the opt-in e2e: half the CDN traffic, no 412s under a lockout CDN, viewer seeks 0.29 s instead of 4 s there. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
428 lines
16 KiB
Dart
428 lines
16 KiB
Dart
import 'dart:async';
|
|
import 'dart:io';
|
|
import 'dart:math';
|
|
import 'dart:typed_data';
|
|
|
|
import 'package:sync_relay_core/core.dart';
|
|
import 'package:sync_relay_server/server.dart';
|
|
import 'package:test/test.dart';
|
|
|
|
// The single-fetch source (SharedSource): one upstream reader per track
|
|
// feeds this app's player (a loopback endpoint) and the upload to the
|
|
// server. Its loopback answers, a cut and a resume, seeks of either
|
|
// reader, the two kept within one window, and the memory bound.
|
|
|
|
const _mib = 1 << 20;
|
|
|
|
/// An in-memory upstream: every read logged by its start, at most
|
|
/// [rate] bytes per second when set.
|
|
class _Upstream implements RelayBytes {
|
|
final Uint8List data;
|
|
final int? rate;
|
|
final reads = <int>[];
|
|
var open = 0, maxOpen = 0, bytes = 0;
|
|
_Upstream(this.data, {this.rate});
|
|
|
|
@override
|
|
Future<int> size() async => data.length;
|
|
|
|
@override
|
|
Stream<List<int>> read(int start, int end) async* {
|
|
reads.add(start);
|
|
open++;
|
|
maxOpen = max(maxOpen, open);
|
|
try {
|
|
const chunk = 64 * 1024;
|
|
for (var pos = start; pos <= end; pos += chunk) {
|
|
final n = min(chunk, end - pos + 1);
|
|
if (rate != null) await Future<void>.delayed(Duration(microseconds: n * 1000000 ~/ rate!));
|
|
bytes += n;
|
|
yield Uint8List.sublistView(data, pos, pos + n);
|
|
}
|
|
} finally {
|
|
open--;
|
|
}
|
|
}
|
|
|
|
@override
|
|
Future<void> close() async {}
|
|
}
|
|
|
|
Uint8List _bytes(int n) => Uint8List.fromList(List.generate(n, (i) => (i * 7 + (i >> 16)) % 251));
|
|
|
|
Future<(HttpClientResponse, List<int>)> _get(String url, {String? range, String method = 'GET'}) async {
|
|
final c = HttpClient();
|
|
try {
|
|
final req = await c.openUrl(method, Uri.parse(url));
|
|
if (range != null) req.headers.set(HttpHeaders.rangeHeader, range);
|
|
final res = await req.close();
|
|
final body = <int>[];
|
|
await for (final b in res) {
|
|
body.addAll(b);
|
|
}
|
|
return (res, body);
|
|
} finally {
|
|
c.close(force: true);
|
|
}
|
|
}
|
|
|
|
Future<List<int>> _readAll(Stream<List<int>> s) async => [for (final c in await s.toList()) ...c];
|
|
|
|
/// The upload's track [i] (what the relay source reads).
|
|
Future<RelayBytes> _upload(SharedSource s, [int i = 0]) async => (await s.upload.open())[i].bytes!;
|
|
|
|
void main() {
|
|
group('loopback endpoint', () {
|
|
late SharedSource src;
|
|
late Uint8List data;
|
|
setUp(() async {
|
|
data = _bytes(3 * _mib + 123);
|
|
src = SharedSource(RelayMedia.single(_Upstream(data)));
|
|
await src.start();
|
|
});
|
|
tearDown(() => src.close());
|
|
|
|
test('answers ranges like the server (200, 206, suffix, 416), connection: close', () async {
|
|
final url = src.playerUrl;
|
|
expect(url, startsWith('http://127.0.0.1:'));
|
|
|
|
var (res, body) = await _get(url);
|
|
expect(res.statusCode, 200);
|
|
expect(res.contentLength, data.length);
|
|
expect(res.headers.value('accept-ranges'), 'bytes');
|
|
expect(res.headers.value('connection'), 'close');
|
|
expect(body, data);
|
|
|
|
(res, body) = await _get(url, range: 'bytes=1000-1999');
|
|
expect(res.statusCode, 206);
|
|
expect(res.headers.value('content-range'), 'bytes 1000-1999/${data.length}');
|
|
expect(body, data.sublist(1000, 2000));
|
|
|
|
(res, body) = await _get(url, range: 'bytes=-50');
|
|
expect(res.statusCode, 206);
|
|
expect(body, data.sublist(data.length - 50));
|
|
|
|
(res, body) = await _get(url, range: 'bytes=${2 * _mib}-');
|
|
expect(res.statusCode, 206);
|
|
expect(body, data.sublist(2 * _mib));
|
|
|
|
(res, body) = await _get(url, range: 'bytes=${data.length}-');
|
|
expect(res.statusCode, 416);
|
|
expect(res.headers.value('content-range'), 'bytes */${data.length}');
|
|
|
|
(res, body) = await _get(url, method: 'HEAD');
|
|
expect(res.statusCode, 200);
|
|
expect(res.contentLength, data.length);
|
|
expect(body, isEmpty);
|
|
});
|
|
|
|
test('a wrong secret or track is 404', () async {
|
|
final u = Uri.parse(src.playerUrl);
|
|
expect((await _get(u.replace(path: '/nope/0').toString())).$1.statusCode, 404);
|
|
expect((await _get(u.replace(path: '${u.pathSegments[0]}/1').toString())).$1.statusCode, 404);
|
|
});
|
|
|
|
test('two tracks: an EDL of one loopback URL per track, no headers', () async {
|
|
final two = SharedSource(_TwoTracks(_Upstream(_bytes(_mib)), _Upstream(_bytes(_mib ~/ 2))));
|
|
addTearDown(two.close);
|
|
await two.open();
|
|
final url = two.playerUrl;
|
|
expect(url, startsWith('edl://'));
|
|
expect(RegExp(r'http://127\.0\.0\.1:\d+/[0-9a-f]+/[01]').allMatches(url).length, 2);
|
|
expect([for (final t in await two.upload.open()) t.kind], ['video', 'audio']);
|
|
expect(await (await _upload(two, 1)).size(), _mib ~/ 2);
|
|
});
|
|
|
|
test('the player can open before the sizes are read: the endpoint waits for them', () async {
|
|
final slow = _SlowSize(_Upstream(data));
|
|
final s = SharedSource(RelayMedia.single(slow));
|
|
addTearDown(s.close);
|
|
await s.open();
|
|
final pending = _get(s.playerUrl, range: 'bytes=0-99');
|
|
await Future<void>.delayed(const Duration(milliseconds: 100));
|
|
slow.go.complete();
|
|
final (res, body) = await pending;
|
|
expect(res.statusCode, 206);
|
|
expect(body, data.sublist(0, 100));
|
|
});
|
|
});
|
|
|
|
group('ResumingBytes', () {
|
|
late HttpServer cdn;
|
|
late Uint8List data;
|
|
// Per request: cut after this many bytes (null: whole), or answer
|
|
// this status.
|
|
late List<Object?> script;
|
|
var requests = 0;
|
|
setUp(() async {
|
|
data = _bytes(_mib + 77);
|
|
requests = 0;
|
|
cdn = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
|
cdn.listen((req) async {
|
|
final step = requests < script.length ? script[requests] : null;
|
|
requests++;
|
|
final m = RegExp(r'bytes=(\d+)-(\d+)').firstMatch(req.headers.value('range') ?? '')!;
|
|
final start = int.parse(m[1]!), end = int.parse(m[2]!);
|
|
if (step is String) {
|
|
req.response.statusCode = int.parse(step);
|
|
await req.response.close();
|
|
return;
|
|
}
|
|
req.response
|
|
..statusCode = 206
|
|
..headers.set('content-range', 'bytes $start-$end/${data.length}')
|
|
..contentLength = end - start + 1;
|
|
final s = await req.response.detachSocket();
|
|
final cut = step as int?;
|
|
if (cut == -1) {
|
|
await s.flush();
|
|
return; // headers only, then silence
|
|
}
|
|
final stop = cut == null ? end + 1 : min(end + 1, start + cut);
|
|
s.add(data.sublist(start, stop));
|
|
await s.flush();
|
|
if (stop <= end) {
|
|
s.destroy();
|
|
} else {
|
|
await s.close();
|
|
}
|
|
});
|
|
});
|
|
tearDown(() => cdn.close(force: true));
|
|
|
|
HttpRelayBytes http() => HttpRelayBytes(Uri.parse('http://127.0.0.1:${cdn.port}/v.mp4'), chunkSize: 0);
|
|
Duration quick(int _) => const Duration(milliseconds: 20);
|
|
|
|
test('a cut goes on from the byte it reached', () async {
|
|
script = [100000, 300000, 5];
|
|
final r = ResumingBytes(http(), backoff: quick);
|
|
expect(await _readAll(r.read(0, data.length - 1)), data);
|
|
expect(r.retries, 3);
|
|
expect(requests, 4);
|
|
expect(r.bytesIn, data.length);
|
|
});
|
|
|
|
test('412, 429 and 5xx back off and try again; 403 and 404 are final', () async {
|
|
script = ['412', '429', '503', null];
|
|
final r = ResumingBytes(http(), backoff: quick);
|
|
expect(await _readAll(r.read(10, 5009)), data.sublist(10, 5010));
|
|
expect(r.retries, 3);
|
|
|
|
for (final status in ['403', '404']) {
|
|
script = [status];
|
|
requests = 0;
|
|
final f = ResumingBytes(http(), backoff: quick);
|
|
await expectLater(_readAll(f.read(0, 99)),
|
|
throwsA(isA<UpstreamHttpException>().having((e) => e.status, 'status', int.parse(status))));
|
|
expect(requests, 1, reason: status);
|
|
}
|
|
});
|
|
|
|
test('an answer that goes silent is cut after the stall limit and resumed', () async {
|
|
script = [-1, null];
|
|
final r = ResumingBytes(http(), backoff: quick, stall: const Duration(milliseconds: 300));
|
|
final sw = Stopwatch()..start();
|
|
expect(await _readAll(r.read(0, 199999)), data.sublist(0, 200000));
|
|
expect(r.retries, 1);
|
|
expect(sw.elapsedMilliseconds, greaterThanOrEqualTo(250));
|
|
});
|
|
|
|
test('cancelling stops it during a pause', () async {
|
|
script = ['503', '503', '503'];
|
|
final r = ResumingBytes(http(), backoff: (_) => const Duration(seconds: 30));
|
|
final sub = r.read(0, 99).listen((_) {});
|
|
await Future<void>.delayed(const Duration(milliseconds: 100));
|
|
final sw = Stopwatch()..start();
|
|
await sub.cancel();
|
|
expect(sw.elapsedMilliseconds, lessThan(1000));
|
|
expect(requests, 1);
|
|
});
|
|
});
|
|
|
|
group('the shared buffer: seeks, one window, memory', () {
|
|
test('cut upstream: both readers get every byte through one resuming reader', () async {
|
|
final data = _bytes(6 * _mib);
|
|
final up = _Cutting(data, every: 700 * 1024);
|
|
final s = SharedSource(RelayMedia.single(up), backoff: (_) => const Duration(milliseconds: 10));
|
|
addTearDown(s.close);
|
|
await s.start();
|
|
final upload = await _upload(s);
|
|
final results = await Future.wait([
|
|
_get(s.playerUrl).then((r) => r.$2),
|
|
_readAll(upload.read(0, data.length - 1)),
|
|
]);
|
|
expect(results[0], data);
|
|
expect(results[1], data);
|
|
expect(s.resumes, greaterThan(3));
|
|
expect(up.maxOpen, 1, reason: 'one upstream reader');
|
|
expect(up.bytes, lessThan(data.length + 3 * _mib), reason: 'each byte read about once');
|
|
});
|
|
|
|
test('the player seeks far ahead and back: the reader follows it, the head is kept', () async {
|
|
final data = _bytes(40 * _mib);
|
|
final up = _Upstream(data);
|
|
final s = SharedSource(RelayMedia.single(up), ahead: 4 * _mib, back: _mib);
|
|
addTearDown(s.close);
|
|
await s.start();
|
|
expect((await _get(s.playerUrl, range: 'bytes=0-${_mib - 1}')).$2, data.sublist(0, _mib));
|
|
final before = up.reads.length;
|
|
final at = 30 * _mib + 12345;
|
|
expect((await _get(s.playerUrl, range: 'bytes=$at-${at + _mib - 1}')).$2, data.sublist(at, at + _mib));
|
|
final seekReads = up.reads.sublist(before);
|
|
expect(seekReads.where((r) => r >= 30 * _mib - RelayConstants.relaySliceBytes && r <= at), isNotEmpty,
|
|
reason: 'a new upstream read at the seek: $seekReads');
|
|
final again = up.reads.length;
|
|
expect((await _get(s.playerUrl, range: 'bytes=1000-1999')).$2, data.sublist(1000, 2000));
|
|
expect(up.reads.sublist(again).where((r) => r < RelayConstants.relayPinBytes), isEmpty,
|
|
reason: 'the head is pinned: no upstream read for it');
|
|
});
|
|
|
|
test('player and upload reading together share one sequential read (no hand-overs)', () async {
|
|
final data = _bytes(24 * _mib);
|
|
final up = _Upstream(data, rate: 60 * _mib);
|
|
final s = SharedSource(RelayMedia.single(up));
|
|
addTearDown(s.close);
|
|
await s.start();
|
|
final upload = await _upload(s);
|
|
final out = await Future.wait([
|
|
_get(s.playerUrl).then((r) => r.$2),
|
|
_readAll(upload.read(0, data.length - 1)),
|
|
]);
|
|
expect(out[0], data);
|
|
expect(out[1], data);
|
|
// The tail pin, the head, then one sequential read (a slice-boundary
|
|
// hand-over between the two would be a new read per 256 KiB).
|
|
expect(up.reads.length, lessThanOrEqualTo(6), reason: '${up.reads}');
|
|
expect(up.bytes, lessThanOrEqualTo(data.length + 2 * RelayConstants.relayPinBytes));
|
|
});
|
|
|
|
test('the upload does not run ahead of a reading player by more than the window; free once the player stops', () async {
|
|
final data = _bytes(48 * _mib);
|
|
final up = _Upstream(data);
|
|
final s = SharedSource(RelayMedia.single(up), ahead: 4 * _mib, back: _mib);
|
|
addTearDown(s.close);
|
|
await s.start();
|
|
final buffer = s.tracks.single.buffer;
|
|
// The player has a request open at 0 and stops reading after its
|
|
// first bytes (mpv with a full cache).
|
|
final player = StreamIterator(buffer.read(SharedSource.playerViewer, 0, data.length - 1));
|
|
expect(await player.moveNext(), isTrue);
|
|
final upload = await _upload(s);
|
|
final far = 30 * _mib;
|
|
final got = <int>[];
|
|
final sub = upload.read(far, far + _mib - 1).listen(got.addAll);
|
|
await Future<void>.delayed(const Duration(milliseconds: 600));
|
|
expect(got, isEmpty, reason: 'held: 30 MiB is past the player + 4 MiB');
|
|
expect(up.reads.where((r) => r >= 20 * _mib && r < data.length - RelayConstants.relayPinBytes), isEmpty,
|
|
reason: 'nothing asked upstream for it: ${up.reads}');
|
|
await player.cancel();
|
|
final until = DateTime.now().add(const Duration(seconds: 5));
|
|
while (got.length < _mib && DateTime.now().isBefore(until)) {
|
|
await Future<void>.delayed(const Duration(milliseconds: 20));
|
|
}
|
|
await sub.cancel();
|
|
expect(got, data.sublist(far, far + _mib));
|
|
});
|
|
|
|
test('the player is held the same way when the upload lags; the tail is always served', () async {
|
|
final data = _bytes(48 * _mib);
|
|
final up = _Upstream(data);
|
|
final s = SharedSource(RelayMedia.single(up), ahead: 4 * _mib, back: _mib);
|
|
addTearDown(s.close);
|
|
await s.start();
|
|
final buffer = s.tracks.single.buffer;
|
|
final upload = StreamIterator((await _upload(s)).read(0, data.length - 1));
|
|
expect(await upload.moveNext(), isTrue);
|
|
// The tail (an mp4 index) is pinned: never held.
|
|
final tail = data.length - 1000;
|
|
expect((await _get(s.playerUrl, range: 'bytes=$tail-')).$2, data.sublist(tail));
|
|
final got = <int>[];
|
|
final sub = buffer.read(SharedSource.playerViewer, 20 * _mib, 21 * _mib - 1).listen(got.addAll);
|
|
await Future<void>.delayed(const Duration(milliseconds: 600));
|
|
expect(got, isEmpty);
|
|
await upload.cancel();
|
|
final until = DateTime.now().add(const Duration(seconds: 5));
|
|
while (got.length < _mib && DateTime.now().isBefore(until)) {
|
|
await Future<void>.delayed(const Duration(milliseconds: 20));
|
|
}
|
|
await sub.cancel();
|
|
expect(got.length, _mib);
|
|
});
|
|
|
|
test('memory stays within the budget while the player reads the whole media', () async {
|
|
final data = _bytes(64 * _mib);
|
|
final up = _Upstream(data);
|
|
const ahead = 8 * _mib, back = 2 * _mib;
|
|
final s = SharedSource(RelayMedia.single(up), ahead: ahead, back: back);
|
|
addTearDown(s.close);
|
|
await s.start();
|
|
final buffer = s.tracks.single.buffer;
|
|
var peak = 0;
|
|
var n = 0;
|
|
await for (final c in buffer.read(SharedSource.playerViewer, 0, data.length - 1)) {
|
|
n += c.length;
|
|
peak = max(peak, buffer.cachedBytes);
|
|
}
|
|
expect(n, data.length);
|
|
expect(peak, lessThanOrEqualTo(ahead + back + 2 * RelayConstants.relayPinBytes + RelayConstants.relaySliceBytes));
|
|
});
|
|
|
|
test('closing under open reads ends them plainly', () async {
|
|
final data = _bytes(8 * _mib);
|
|
final s = SharedSource(RelayMedia.single(_Upstream(data, rate: _mib)));
|
|
await s.start();
|
|
final upload = await _upload(s);
|
|
final errors = <Object>[];
|
|
final sub = upload.read(0, data.length - 1).listen((_) {}, onError: errors.add);
|
|
final player = _get(s.playerUrl).then((r) => r.$2.length, onError: (_) => -1);
|
|
await Future<void>.delayed(const Duration(milliseconds: 200));
|
|
await s.close();
|
|
await player;
|
|
await Future<void>.delayed(const Duration(milliseconds: 50));
|
|
await sub.cancel();
|
|
expect(errors.every((e) => e is RelayEnded), isTrue, reason: '$errors');
|
|
});
|
|
});
|
|
}
|
|
|
|
class _TwoTracks implements RelayMedia {
|
|
final RelayBytes v, a;
|
|
_TwoTracks(this.v, this.a);
|
|
@override
|
|
Future<List<RelayTrack>> open() async => [RelayTrack('video', v), RelayTrack('audio', a)];
|
|
@override
|
|
Future<void> close() async {}
|
|
}
|
|
|
|
class _SlowSize implements RelayBytes {
|
|
final RelayBytes inner;
|
|
final go = Completer<void>();
|
|
_SlowSize(this.inner);
|
|
@override
|
|
Future<int> size() async {
|
|
await go.future;
|
|
return inner.size();
|
|
}
|
|
|
|
@override
|
|
Stream<List<int>> read(int start, int end) => inner.read(start, end);
|
|
@override
|
|
Future<void> close() => inner.close();
|
|
}
|
|
|
|
/// An upstream whose every read is cut after [every] bytes.
|
|
class _Cutting extends _Upstream {
|
|
final int every;
|
|
_Cutting(super.data, {required this.every});
|
|
|
|
@override
|
|
Stream<List<int>> read(int start, int end) async* {
|
|
var n = 0;
|
|
await for (final c in super.read(start, end)) {
|
|
if (n + c.length > every) throw const HttpException('Connection closed while receiving data');
|
|
n += c.length;
|
|
yield c;
|
|
}
|
|
}
|
|
}
|