perf(mobile): sign relay queries off the UI isolate

Signed-off-by: Tom Brow <tomb@block.xyz>
This commit is contained in:
Tom Brow
2026-10-18 17:02:59 -07:00
parent 8f2c7710e9
commit 5fa77682eb
4 changed files with 227 additions and 10 deletions

View File

@@ -76,6 +76,7 @@ class _BufferedEvent {
class RelaySessionNotifier extends Notifier<SessionState> {
RelaySessionNotifier({
http.Client? httpClient,
RelayQuerySigner? querySigner,
http.Client Function()? httpClientFactory,
RelaySocketFactory socketFactory = RelaySocket.new,
DateTime Function()? now,
@@ -86,6 +87,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
client: httpClient,
clientFactory: httpClientFactory,
),
_querySigner = querySigner ?? RelayQuerySigner(),
_socketFactory = socketFactory,
_now = now ?? DateTime.now,
_rateLimitGate = rateLimitGate ?? RelayRateLimitGate(),
@@ -93,6 +95,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
_replayDelay = replayDelay;
final RelayHttpQueryClient _httpQueryClient;
final RelayQuerySigner _querySigner;
final RelaySocketFactory _socketFactory;
final DateTime Function() _now;
final RelayRateLimitGate _rateLimitGate;
@@ -233,17 +236,21 @@ class RelaySessionNotifier extends Notifier<SessionState> {
final config = ref.read(relayConfigProvider);
final url = Uri.parse(config.baseUrl).resolve(path).toString();
final bodyBytes = utf8.encode(jsonEncode(payload));
final authorization = await _querySigner.sign((
method: 'POST',
url: url,
bodyBytes: bodyBytes,
nsec: config.nsec,
));
if (_disposed || _ageRestricted || generation != _contextGeneration) {
throw StateError('Relay query belongs to a retired session');
}
// Reuse the session transport on success. A timeout rotates immediately
// for new queries, then closes the retired client after its peers finish.
final response = await _httpQueryClient.post(
Uri.parse(url),
headers: {
'Authorization': buildNip98AuthHeader(
method: 'POST',
url: url,
bodyBytes: bodyBytes,
nsec: config.nsec,
),
'Authorization': authorization,
'Content-Type': 'application/json',
},
body: bodyBytes,
@@ -1041,6 +1048,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
void _dispose() {
_disposed = true;
_contextGeneration++;
_querySigner.retire();
_beforePauseCallbacks.clear();
_connectionGeneration++;
_reconnectTimer?.cancel();

View File

@@ -1,5 +1,71 @@
part of 'relay_session.dart';
/// Inputs passed to a query-auth signing worker.
typedef RelayQueryAuthRequest = ({
String method,
String url,
List<int> bodyBytes,
String? nsec,
});
/// Serializes query authentication off the UI isolate, with bounded admission.
///
/// Proofs are minted when their job starts, not while waiting in the queue.
/// Retiring the session cancels queued jobs and discards active results.
class RelayQuerySigner {
/// Creates a signer; tests may supply a controlled asynchronous worker.
RelayQuerySigner({
@visibleForTesting Future<String> Function(RelayQueryAuthRequest)? sign,
}) : _sign = sign ?? _signQueryInBackground;
final Future<String> Function(RelayQueryAuthRequest) _sign;
static const _maxPending = 64;
Future<void> _tail = Future<void>.value();
int _pending = 0;
int _generation = 0;
/// Signs one request, rejecting excess work without retaining another job.
Future<String> sign(RelayQueryAuthRequest request) {
if (_pending >= _maxPending) {
return Future.error(StateError('Relay query signing queue is full'));
}
final generation = _generation;
_pending++;
final released = Completer<void>();
final result = _tail.then((_) async {
try {
if (generation != _generation) {
throw StateError('Relay query belongs to a retired session');
}
final header = await _sign(request);
if (generation != _generation) {
throw StateError('Relay query belongs to a retired session');
}
return header;
} finally {
_pending--;
released.complete();
}
});
// Failure is delivered to the caller; it must not poison later jobs.
_tail = released.future;
return result;
}
/// Invalidates admitted work while allowing a rebuilt session to sign again.
void retire() => _generation++;
}
Future<String> _signQueryInBackground(RelayQueryAuthRequest request) =>
compute(_signQuery, request, debugLabel: 'relay-query-auth');
String _signQuery(RelayQueryAuthRequest request) => buildNip98AuthHeader(
method: request.method,
url: request.url,
bodyBytes: request.bodyBytes,
nsec: request.nsec,
);
String buildNip98AuthHeader({
required String method,
required String url,

View File

@@ -0,0 +1,74 @@
import 'dart:async';
import 'package:buzz/shared/relay/relay.dart';
import 'package:flutter_test/flutter_test.dart';
const _request = (
method: 'POST',
url: 'https://relay.example/query',
bodyBytes: <int>[91, 93],
nsec: 'test-key',
);
void main() {
test(
'serializes workers and rejects saturation without retaining a job',
() async {
final worker = Completer<String>();
var calls = 0;
final signer = RelayQuerySigner(
sign: (_) {
calls++;
return worker.future;
},
);
final requests = List.generate(64, (_) => signer.sign(_request));
await expectLater(signer.sign(_request), throwsStateError);
expect(calls, 1);
worker.complete('signed');
expect(await Future.wait(requests), everyElement('signed'));
expect(calls, 64);
expect(await signer.sign(_request), 'signed');
},
);
test(
'retirement cancels queued work and rejects the active result',
() async {
final worker = Completer<String>();
var calls = 0;
final signer = RelayQuerySigner(
sign: (_) {
calls++;
return worker.future;
},
);
final requests = List.generate(
3,
(_) => expectLater(signer.sign(_request), throwsStateError),
);
await pumpEventQueue();
expect(calls, 1);
signer.retire();
worker.complete('retired proof');
await Future.wait(requests);
expect(calls, 1);
expect(await signer.sign(_request), 'retired proof');
expect(calls, 2);
},
);
test('a failed worker does not poison the next request', () async {
var calls = 0;
final signer = RelayQuerySigner(
sign: (_) async {
if (++calls == 1) throw const FormatException('invalid signing key');
return 'signed';
},
);
final failed = expectLater(signer.sign(_request), throwsFormatException);
final next = signer.sign(_request);
await failed;
expect(await next, 'signed');
});
}

View File

@@ -14,6 +14,66 @@ import 'package:buzz/shared/auth/auth_provider.dart';
import 'package:buzz/shared/relay/relay.dart';
void main() {
for (final retirement in ['community', 'age', 'dispose']) {
test(
'query retired during signing sends no HTTP request ($retirement)',
() async {
final started = Completer<void>();
final proof = Completer<String>();
final age = _MutableAgeNotifier();
var requests = 0;
final session = RelaySessionNotifier(
querySigner: RelayQuerySigner(
sign: (_) {
started.complete();
return proof.future;
},
),
httpClient: http_testing.MockClient((_) async {
requests++;
return http.Response('[]', 200);
}),
);
final container = ProviderContainer(
overrides: [
authProvider.overrideWith(_PendingAuthNotifier.new),
ageSignalProvider.overrideWith(() => age),
relayConfigProvider.overrideWith(
() => _FakeRelayConfigNotifier(
baseUrl: 'https://relay.example',
nsec: 'test-key',
),
),
relaySessionProvider.overrideWith(() => session),
],
);
addTearDown(container.dispose);
container.read(relaySessionProvider);
final query = expectLater(
session.queryRelay(const []),
throwsStateError,
);
await started.future;
expect(requests, 0);
switch (retirement) {
case 'community':
container
.read(relayConfigProvider.notifier)
.update(baseUrl: 'https://other.example', nsec: 'other-key');
container.read(relaySessionProvider);
case 'age':
age.setState(AgeSignalState.restricted);
container.read(relaySessionProvider);
case 'dispose':
session.debugDispose();
}
proof.complete('Nostr stale');
await query;
expect(requests, 0);
},
);
}
test(
'confirmed minor tears down retained relay and observer providers',
() async {
@@ -23,6 +83,7 @@ void main() {
var httpCalls = 0;
final pendingResponse = Completer<http.Response>();
final session = RelaySessionNotifier(
querySigner: RelayQuerySigner(sign: (_) async => 'Nostr test'),
httpClient: http_testing.MockClient((_) {
httpCalls++;
return pendingResponse.future;
@@ -196,6 +257,7 @@ void main() {
.map((byte) => byte.toRadixString(16).padLeft(2, '0'))
.join();
expect(nostr.Event.fromMap(authEvent).isValid(), isTrue);
expect(authEvent['kind'], 27235);
expect(authEvent['pubkey'], keychain.public);
expect(
@@ -237,6 +299,7 @@ void main() {
test('queryRelay rotates the client after a timeout', () async {
final clients = <_ControlledHttpClient>[];
final session = RelaySessionNotifier(
querySigner: RelayQuerySigner(sign: (_) async => 'Nostr test'),
httpClientFactory: () {
final client = _ControlledHttpClient();
clients.add(client);
@@ -265,6 +328,7 @@ void main() {
expect(clients.single.closed, isTrue);
final nextQuery = session.queryRelay(const []);
await pumpEventQueue();
expect(clients, hasLength(2));
clients.last.complete(http.Response('[]', 200));
@@ -277,6 +341,7 @@ void main() {
() async {
final clients = <_QueuedControlledHttpClient>[];
final session = RelaySessionNotifier(
querySigner: RelayQuerySigner(sign: (_) async => 'Nostr test'),
httpClientFactory: () {
final client = _QueuedControlledHttpClient();
clients.add(client);
@@ -304,12 +369,18 @@ void main() {
timeout: const Duration(milliseconds: 10),
);
final peerQuery = session.queryRelay(const []);
final timedOut = expectLater(
timedOutQuery,
throwsA(isA<TimeoutException>()),
);
await pumpEventQueue();
expect(clients.single.requestCount, 2);
await expectLater(timedOutQuery, throwsA(isA<TimeoutException>()));
await timedOut;
expect(clients.single.closed, isFalse);
final nextQuery = session.queryRelay(const []);
await pumpEventQueue();
expect(clients, hasLength(2));
clients.first.complete(1, http.Response('[]', 200));
expect(await peerQuery, isEmpty);
@@ -485,10 +556,8 @@ void main() {
expect(gate.isActive, isTrue);
final query = harness.session.queryRelay(const []);
await Future<void>.delayed(Duration.zero);
expect(requestCount, 1);
expect(await query, isEmpty);
expect(requestCount, 1);
// Still armed: the read must neither wait on the gate nor clear it.
expect(gate.isActive, isTrue);
});