Retry mounted media once after signing queue drains

Signed-off-by: Tom Brow <tomb@block.xyz>
This commit is contained in:
Tom Brow
2026-10-18 13:58:18 -07:00
committed by Tom Brow
parent b6dd9bca45
commit bfb0f52d9b
5 changed files with 215 additions and 5 deletions

View File

@@ -33,7 +33,7 @@ final videoPreviewFrameLoaderProvider = Provider<VideoPreviewFrameLoader>((
return (url) async {
final Map<String, String> headers;
try {
headers = await auth.headersForAsync(url);
headers = await auth.headersForAsyncWithRetry(url);
} catch (_) {
// Queue rejection or a failed signing worker leaves this preview
// unavailable. Do not attempt an unauthenticated video request.

View File

@@ -18,6 +18,16 @@ const _mediaGetAuthLifetimeSeconds = 60;
/// mints a fresh token (mint-per-request pattern for NIP-FI compliance).
const _mediaGetAuthRefreshMarginSeconds = 60;
/// Queue saturation with a shared signal for the already-admitted work to drain.
class MediaAuthQueueFull extends StateError {
/// Completes when the queue snapshot that rejected this request has drained.
final Future<void> drained;
/// Retains only the existing queue tail, not another signing job.
MediaAuthQueueFull(this.drained)
: super('Media authentication queue is full');
}
/// Builds BUD-01 Blossom `t=get` auth headers for relay-host media URLs.
///
/// Returns an empty map for non-relay URLs or when no signing key is available,
@@ -66,15 +76,16 @@ class MediaGetAuthService {
/// Image loading starts during layout. Serialize signing per auth scope so a
/// viewport full of uncached images does not spawn an isolate per row at once.
/// At most 64 requests may be outstanding. Excess requests fail explicitly
/// without spawning or retaining work; image failures are evicted so a later
/// resolution can retry. Timestamp proofs when their queued jobs start.
/// without spawning or retaining a signing job. Consumers can use
/// [headersForAsyncWithRetry] for one retry after the queue drains.
/// Timestamp proofs when their queued jobs start.
Future<Map<String, String>> headersForAsync(String url) {
final nsec = _nsec;
if (nsec == null || nsec.isEmpty || !isRelayMediaUrl(url)) {
return Future.value(const {});
}
if (_pendingSignatures >= _maxPendingSignatures) {
return Future.error(StateError('Media authentication queue is full'));
return Future.error(MediaAuthQueueFull(_signingTail));
}
_pendingSignatures++;
final released = Completer<void>();
@@ -97,6 +108,19 @@ class MediaGetAuthService {
return result;
}
/// Keeps an in-flight consumer pending through one queue-saturation retry.
///
/// Waits for the existing queue snapshot, then competes for normal admission.
/// A second rejection propagates; no retry loop or extra signing queue exists.
Future<Map<String, String>> headersForAsyncWithRetry(String url) async {
try {
return await headersForAsync(url);
} on MediaAuthQueueFull catch (error) {
await error.drained;
return headersForAsync(url);
}
}
Map<String, String> headersFor(String url) {
final nsec = _nsec;
if (nsec == null || nsec.isEmpty) return const {};

View File

@@ -101,7 +101,7 @@ class MediaImageProvider extends ImageProvider<MediaImageProvider> {
}
final uri = Uri.parse(url);
final headers = await auth.headersForAsync(url);
final headers = await auth.headersForAsyncWithRetry(url);
final http.Response response;
try {
final abort = Completer<void>();

View File

@@ -0,0 +1,168 @@
import 'dart:async';
import 'dart:convert';
import 'package:buzz/features/channels/message_content.dart';
import 'package:buzz/shared/relay/media_auth.dart';
import 'package:buzz/shared/relay/media_image.dart';
import 'package:buzz/shared/theme/theme.dart';
import 'package:flutter/material.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:http/http.dart' as http;
import 'package:http/testing.dart';
import 'package:video_player_platform_interface/video_player_platform_interface.dart';
class _SaturatedAuth extends MediaGetAuthService {
_SaturatedAuth({this.alwaysReject = false})
: super(baseUrl: 'https://example.com', nsec: null);
final bool alwaysReject;
final drained = Completer<void>();
int calls = 0;
@override
Future<Map<String, String>> headersForAsync(String url) async {
calls++;
if (calls == 1 || alwaysReject) throw MediaAuthQueueFull(drained.future);
return {'Authorization': 'test-proof'};
}
}
class _VideoPlatform extends VideoPlayerPlatform {
final sources = <DataSource>[];
@override
Future<void> init() async {}
@override
Future<int?> createWithOptions(VideoCreationOptions options) async {
sources.add(options.dataSource);
return 1;
}
@override
Stream<VideoEvent> videoEventsFor(int playerId) => Stream.value(
VideoEvent(
eventType: VideoEventType.initialized,
size: const Size(100, 100),
duration: const Duration(seconds: 1),
),
);
@override
Widget buildView(int playerId) => const SizedBox();
@override
Future<void> dispose(int playerId) async {}
@override
Future<void> pause(int playerId) async {}
@override
Future<void> setLooping(int playerId, bool looping) async {}
@override
Future<void> setVolume(int playerId, double volume) async {}
@override
Future<void> seekTo(int playerId, Duration position) async {}
@override
Future<void> setMixWithOthers(bool mixWithOthers) async {}
}
void main() {
test('retries saturation only once', () async {
final auth = _SaturatedAuth(alwaysReject: true);
final result = auth.headersForAsyncWithRetry('https://example.com/media/a');
final failure = expectLater(result, throwsA(isA<MediaAuthQueueFull>()));
await Future<void>.delayed(Duration.zero);
expect(auth.calls, 1);
auth.drained.complete();
await failure;
expect(auth.calls, 2);
});
testWidgets('mounted image succeeds after signing capacity drains', (
tester,
) async {
final auth = _SaturatedAuth();
final png = base64Decode(
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=',
);
var fetches = 0;
final client = MockClient((request) async {
fetches++;
expect(request.headers['Authorization'], 'test-proof');
return http.Response.bytes(png, 200);
});
await tester.pumpWidget(
MaterialApp(
home: Image(
image: MediaImageProvider(
url: 'https://example.com/media/recovery.png',
auth: auth,
client: client,
),
errorBuilder: (_, _, _) => const Text('failed'),
),
),
);
await tester.pump();
final element = tester.element(find.byType(Image));
expect(auth.calls, 1);
expect(fetches, 0);
auth.drained.complete();
await tester.runAsync(() async {
await Future<void>.delayed(const Duration(milliseconds: 100));
});
await tester.pumpAndSettle();
expect(tester.element(find.byType(Image)), same(element));
expect(auth.calls, 2);
expect(fetches, 1);
expect(tester.widget<RawImage>(find.byType(RawImage)).image, isNotNull);
expect(find.text('failed'), findsNothing);
await tester.pumpWidget(const SizedBox());
PaintingBinding.instance.imageCache.clear();
PaintingBinding.instance.imageCache.clearLiveImages();
client.close();
});
testWidgets(
'mounted posterless video succeeds after signing capacity drains',
(tester) async {
const url = 'https://example.com/media/recovery.mp4';
final auth = _SaturatedAuth();
final previous = VideoPlayerPlatform.instance;
final platform = _VideoPlatform();
VideoPlayerPlatform.instance = platform;
addTearDown(() => VideoPlayerPlatform.instance = previous);
await tester.pumpWidget(
ProviderScope(
overrides: [mediaGetAuthServiceProvider.overrideWithValue(auth)],
child: MaterialApp(
theme: AppTheme.light(),
home: const AppMarkdownTheme(
child: Scaffold(
body: MessageContent(
content: '![video]($url)',
tags: [
['imeta', 'url $url', 'm video/mp4'],
],
),
),
),
),
),
);
await tester.pump();
final element = tester.element(find.byType(MessageContent));
expect(auth.calls, 1);
expect(platform.sources, isEmpty);
auth.drained.complete();
await tester.pumpAndSettle();
expect(tester.element(find.byType(MessageContent)), same(element));
expect(auth.calls, 2);
expect(
platform.sources.single.httpHeaders['Authorization'],
'test-proof',
);
expect(
find.byKey(const ValueKey('message-media-video-first-frame:$url')),
findsOneWidget,
);
expect(tester.takeException(), isNull);
await tester.pumpWidget(const SizedBox());
await tester.pump();
},
);
}

View File

@@ -126,6 +126,24 @@ void main() {
expect(await auth.headersForAsync(_mediaUrl), isNotEmpty);
});
test('retry bursts retain the 64-job admission limit', () async {
final auth = _auth(nsec: nostr.Keys.generate().nsec);
final requests = List.generate(
160,
(_) => auth
.headersForAsyncWithRetry(_mediaUrl)
.then<Object>(
(headers) => headers,
onError: (Object error) => error,
),
);
final results = await Future.wait(requests);
// 64 initial admissions, then 64 retry admissions; the remaining
// consumers fail their second admission instead of growing the queue.
expect(results.whereType<Map<String, String>>(), hasLength(128));
expect(results.whereType<MediaAuthQueueFull>(), hasLength(32));
});
test('keeps the signing identity separate across auth scopes', () async {
final first = nostr.Keys.generate();
final second = nostr.Keys.generate();