edde746
304d8a5e33
A relay connection can win the phase-1 race (plex.tv edge answers fast), gets saved as the preferred endpoint, and the cached URL then wins the deterministic head-start probe at every bind — pinning future sessions to relay's 2 Mbps cap even after direct connectivity returns, surfacing as HTTP 500s on Original quality. A session that lands on relay legitimately (direct down at connect time) stays there all evening: re-optimization only fired on connectivity events. Relay is now a fallback tier in the race: all probes still start at once, but a relay success is held until every direct candidate has failed, and a cached relay URL gets no head start. A single persist gate refuses relay at all four endpoint-save sites, classified against both the live server and the connect-time capture so rotated relay URIs cannot slip through. While the active endpoint is relay, a bounded-backoff re-probe (30s/60s/ 120s) re-races the candidates so a returning direct endpoint promotes away without a restart.
427 lines
14 KiB
Dart
427 lines
14 KiB
Dart
import 'dart:async';
|
|
import 'package:fake_async/fake_async.dart';
|
|
|
|
import 'package:flutter_test/flutter_test.dart';
|
|
import 'package:plezy/utils/app_logger.dart';
|
|
import 'package:plezy/utils/endpoint_race.dart';
|
|
import 'package:plezy/utils/log_redaction_manager.dart';
|
|
|
|
typedef _Result = ({String url, bool ok});
|
|
|
|
void main() {
|
|
const headStart = Duration(milliseconds: 60);
|
|
setUp(() {
|
|
MemoryLogOutput.clearLogs();
|
|
LogRedactionManager.clearTrackedValues();
|
|
setLoggerLevel(true);
|
|
});
|
|
tearDown(() {
|
|
MemoryLogOutput.clearLogs();
|
|
LogRedactionManager.clearTrackedValues();
|
|
});
|
|
|
|
Stream<EndpointRaceSelection<String, _Result>> race({
|
|
required List<String> candidates,
|
|
String? preferred,
|
|
int Function(String url)? tierOf,
|
|
required Future<_Result> Function(String url) probe,
|
|
Future<_Result> Function(String url)? measure,
|
|
String? Function(Map<String, _Result> results)? selectBest,
|
|
Map<String, Object?> Function(String candidate, _Result result)? failureLogFields,
|
|
}) {
|
|
return raceEndpointCandidates<String, _Result>(
|
|
label: 'test',
|
|
candidates: candidates,
|
|
urlOf: (c) => c,
|
|
preferredUrl: preferred,
|
|
tierOf: tierOf,
|
|
failureLogFields: failureLogFields,
|
|
probe: (c, _) => probe(c),
|
|
measure: measure ?? (c) async => (url: c, ok: false),
|
|
isSuccess: (r) => r.ok,
|
|
selectBestCandidate: selectBest ?? (results) => results.keys.first,
|
|
preferredTimeout: const Duration(milliseconds: 500),
|
|
preferredHeadStart: headStart,
|
|
raceTimeout: const Duration(milliseconds: 500),
|
|
);
|
|
}
|
|
|
|
test('preferred endpoint diagnostics contain no candidate literals', () async {
|
|
const canary = 'https://preferred-race-canary.invalid/private-race-path';
|
|
|
|
final selections = await race(
|
|
candidates: const [canary],
|
|
preferred: canary,
|
|
probe: (url) async => (url: url, ok: true),
|
|
measure: (url) async => (url: url, ok: true),
|
|
).toList();
|
|
final storedFields = MemoryLogOutput.getLogs().expand<String>(
|
|
(entry) => [entry.message, if (entry.error != null) entry.error.toString()],
|
|
);
|
|
|
|
expect(selections.map((selection) => selection.candidate), everyElement(canary));
|
|
for (final field in storedFields) {
|
|
expect(field, isNot(contains('preferred-race-canary.invalid')));
|
|
expect(field, isNot(contains('private-race-path')));
|
|
}
|
|
});
|
|
|
|
test('candidate failure diagnostics sanitize endpoint-bearing fields', () async {
|
|
const canary = 'https://failure-race-canary.invalid/private-failure-path';
|
|
|
|
final selections = await race(
|
|
candidates: const [canary],
|
|
probe: (url) async => (url: url, ok: false),
|
|
failureLogFields: (candidate, _) => {
|
|
'error': 'probe failed at $candidate on failure-race-canary.invalid path /private-failure-path',
|
|
},
|
|
).toList();
|
|
final storedFields = MemoryLogOutput.getLogs().expand<String>(
|
|
(entry) => [entry.message, if (entry.error != null) entry.error.toString()],
|
|
);
|
|
|
|
expect(selections, isEmpty);
|
|
for (final field in storedFields) {
|
|
expect(field, isNot(contains('failure-race-canary.invalid')));
|
|
expect(field, isNot(contains('private-failure-path')));
|
|
}
|
|
});
|
|
|
|
test('healthy cached endpoint wins within the head start without racing', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
late Completer<_Result> cachedGate;
|
|
|
|
race(
|
|
candidates: ['a', 'cached'],
|
|
preferred: 'cached',
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
final gate = Completer<_Result>();
|
|
if (url == 'cached') cachedGate = gate;
|
|
return gate.future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
expect(probeCounts, {'cached': 1});
|
|
cachedGate.complete((url: 'cached', ok: true));
|
|
async.flushMicrotasks();
|
|
|
|
expect(selections.first.phase, EndpointRacePhase.first);
|
|
expect(selections.first.candidate, 'cached');
|
|
expect(selections.first.fromPreferred, isTrue);
|
|
expect(probeCounts.containsKey('a'), isFalse);
|
|
|
|
async.elapse(headStart);
|
|
async.flushMicrotasks();
|
|
});
|
|
});
|
|
|
|
test('stale-slow cached endpoint overlaps the race instead of serially blocking it', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
late Completer<_Result> cachedGate;
|
|
late Completer<_Result> fastGate;
|
|
|
|
race(
|
|
candidates: ['fast', 'cached'],
|
|
preferred: 'cached',
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
final gate = Completer<_Result>();
|
|
if (url == 'cached') {
|
|
cachedGate = gate;
|
|
} else {
|
|
fastGate = gate;
|
|
}
|
|
return gate.future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
expect(probeCounts, {'cached': 1});
|
|
async.elapse(headStart - const Duration(milliseconds: 1));
|
|
async.flushMicrotasks();
|
|
expect(probeCounts, {'cached': 1});
|
|
|
|
async.elapse(const Duration(milliseconds: 1));
|
|
async.flushMicrotasks();
|
|
expect(probeCounts, {'cached': 1, 'fast': 1});
|
|
|
|
fastGate.complete((url: 'fast', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections.first.phase, EndpointRacePhase.first);
|
|
expect(selections.first.candidate, 'fast');
|
|
expect(selections.first.fromPreferred, isFalse);
|
|
expect(probeCounts['cached'], 1);
|
|
|
|
cachedGate.complete((url: 'cached', ok: false));
|
|
async.flushMicrotasks();
|
|
});
|
|
});
|
|
|
|
test('cached endpoint that answers after the head start still wins when first', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
late Completer<_Result> cachedGate;
|
|
late Completer<_Result> slowGate;
|
|
|
|
race(
|
|
candidates: ['slow', 'cached'],
|
|
preferred: 'cached',
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
final gate = Completer<_Result>();
|
|
if (url == 'cached') {
|
|
cachedGate = gate;
|
|
} else {
|
|
slowGate = gate;
|
|
}
|
|
return gate.future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
expect(probeCounts, {'cached': 1});
|
|
async.elapse(headStart);
|
|
async.flushMicrotasks();
|
|
expect(probeCounts, {'cached': 1, 'slow': 1});
|
|
|
|
cachedGate.complete((url: 'cached', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections.first.candidate, 'cached');
|
|
expect(selections.first.fromPreferred, isTrue);
|
|
expect(probeCounts['cached'], 1);
|
|
|
|
slowGate.complete((url: 'slow', ok: true));
|
|
async.flushMicrotasks();
|
|
});
|
|
});
|
|
|
|
test('cached endpoint failing within the head start falls back to a fresh race', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final cachedGates = <Completer<_Result>>[];
|
|
late Completer<_Result> altGate;
|
|
|
|
race(
|
|
candidates: ['cached', 'alt'],
|
|
preferred: 'cached',
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
final gate = Completer<_Result>();
|
|
if (url == 'cached') {
|
|
cachedGates.add(gate);
|
|
} else {
|
|
altGate = gate;
|
|
}
|
|
return gate.future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
cachedGates.single.complete((url: 'cached', ok: false));
|
|
async.flushMicrotasks();
|
|
expect(probeCounts, {'cached': 2, 'alt': 1});
|
|
|
|
cachedGates.last.complete((url: 'cached', ok: false));
|
|
altGate.complete((url: 'alt', ok: true));
|
|
async.flushMicrotasks();
|
|
|
|
expect(selections.first.candidate, 'alt');
|
|
expect(selections.first.fromPreferred, isFalse);
|
|
expect(probeCounts['cached'], 2);
|
|
|
|
async.elapse(headStart);
|
|
async.flushMicrotasks();
|
|
});
|
|
});
|
|
|
|
test('preferred URL not among candidates skips the cached probe entirely', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final gates = <String, Completer<_Result>>{};
|
|
|
|
race(
|
|
candidates: ['a', 'b'],
|
|
preferred: 'custom-url',
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
return (gates[url] = Completer<_Result>()).future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
expect(probeCounts.containsKey('custom-url'), isFalse);
|
|
gates['a']!.complete((url: 'a', ok: true));
|
|
gates['b']!.complete((url: 'b', ok: false));
|
|
async.flushMicrotasks();
|
|
|
|
expect(selections.first.candidate, 'a');
|
|
expect(selections.first.fromPreferred, isFalse);
|
|
});
|
|
});
|
|
|
|
test('emits nothing when every candidate fails', () async {
|
|
final selections = await race(
|
|
candidates: ['a', 'b'],
|
|
preferred: 'a',
|
|
probe: (url) async => (url: url, ok: false),
|
|
).toList();
|
|
|
|
expect(selections, isEmpty);
|
|
});
|
|
|
|
test('phase 2 still promotes the selector-best endpoint', () {
|
|
fakeAsync((async) {
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final gates = <String, Completer<_Result>>{};
|
|
|
|
race(
|
|
candidates: ['quick', 'better'],
|
|
probe: (url) => (gates[url] = Completer<_Result>()).future,
|
|
measure: (url) async => (url: url, ok: true),
|
|
selectBest: (results) => 'better',
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
gates['quick']!.complete((url: 'quick', ok: true));
|
|
async.flushMicrotasks();
|
|
gates['better']!.complete((url: 'better', ok: true));
|
|
async.flushMicrotasks();
|
|
|
|
expect(selections, hasLength(2));
|
|
expect(selections.first.phase, EndpointRacePhase.first);
|
|
expect(selections.first.candidate, 'quick');
|
|
expect(selections.last.phase, EndpointRacePhase.best);
|
|
expect(selections.last.candidate, 'better');
|
|
});
|
|
});
|
|
|
|
group('fallback tiers', () {
|
|
int relayTier(String url) => url == 'relay' ? 1 : 0;
|
|
|
|
test('fallback-tier success is held and accepted only after preferred tiers fail', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final gates = <String, Completer<_Result>>{};
|
|
|
|
race(
|
|
candidates: ['direct', 'relay'],
|
|
tierOf: relayTier,
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
return (gates[url] = Completer<_Result>()).future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
// The hold delays acceptance, never the probe: both start at once.
|
|
expect(probeCounts, {'direct': 1, 'relay': 1});
|
|
|
|
gates['relay']!.complete((url: 'relay', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections, isEmpty);
|
|
|
|
gates['direct']!.complete((url: 'direct', ok: false));
|
|
async.flushMicrotasks();
|
|
expect(selections.first.phase, EndpointRacePhase.first);
|
|
expect(selections.first.candidate, 'relay');
|
|
});
|
|
});
|
|
|
|
test('a slower preferred-tier success beats an earlier fallback-tier success', () {
|
|
fakeAsync((async) {
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final gates = <String, Completer<_Result>>{};
|
|
|
|
race(
|
|
candidates: ['direct', 'relay'],
|
|
tierOf: relayTier,
|
|
probe: (url) => (gates[url] = Completer<_Result>()).future,
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
gates['relay']!.complete((url: 'relay', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections, isEmpty);
|
|
|
|
gates['direct']!.complete((url: 'direct', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections.first.candidate, 'direct');
|
|
});
|
|
});
|
|
|
|
test('fallback tier wins without delay once every preferred-tier candidate has failed', () {
|
|
fakeAsync((async) {
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final gates = <String, Completer<_Result>>{};
|
|
|
|
race(
|
|
candidates: ['direct', 'relay'],
|
|
tierOf: relayTier,
|
|
probe: (url) => (gates[url] = Completer<_Result>()).future,
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
gates['direct']!.complete((url: 'direct', ok: false));
|
|
async.flushMicrotasks();
|
|
gates['relay']!.complete((url: 'relay', ok: true));
|
|
async.flushMicrotasks();
|
|
|
|
// No hold window remains — accepted on the spot, no timer elapse.
|
|
expect(selections.first.candidate, 'relay');
|
|
});
|
|
});
|
|
|
|
test('cached fallback-tier URL gets no head start and still loses to a preferred tier', () {
|
|
fakeAsync((async) {
|
|
final probeCounts = <String, int>{};
|
|
final selections = <EndpointRaceSelection<String, _Result>>[];
|
|
final gates = <String, Completer<_Result>>{};
|
|
|
|
race(
|
|
candidates: ['direct', 'relay'],
|
|
preferred: 'relay',
|
|
tierOf: relayTier,
|
|
probe: (url) {
|
|
probeCounts[url] = (probeCounts[url] ?? 0) + 1;
|
|
return (gates[url] = Completer<_Result>()).future;
|
|
},
|
|
).listen(selections.add);
|
|
async.flushMicrotasks();
|
|
|
|
// A cached relay URL (persisted by an older build) is demoted to an
|
|
// ordinary participant: no head-start window where only it probes.
|
|
expect(probeCounts, {'direct': 1, 'relay': 1});
|
|
|
|
gates['relay']!.complete((url: 'relay', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections, isEmpty);
|
|
|
|
gates['direct']!.complete((url: 'direct', ok: true));
|
|
async.flushMicrotasks();
|
|
expect(selections.first.candidate, 'direct');
|
|
expect(selections.first.fromPreferred, isFalse);
|
|
});
|
|
});
|
|
|
|
test('emits nothing when every tier fails', () async {
|
|
final selections = await race(
|
|
candidates: ['direct', 'relay'],
|
|
tierOf: relayTier,
|
|
probe: (url) async => (url: url, ok: false),
|
|
).toList();
|
|
|
|
expect(selections, isEmpty);
|
|
});
|
|
});
|
|
}
|