From 5fa77682ebf4f430eab5cc155a9d8420daae85da Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Sun, 18 Oct 2026 17:02:59 -0700 Subject: [PATCH] perf(mobile): sign relay queries off the UI isolate Signed-off-by: Tom Brow --- mobile/lib/shared/relay/relay_session.dart | 20 +++-- .../lib/shared/relay/relay_session_auth.dart | 66 ++++++++++++++++ .../shared/relay/relay_query_signer_test.dart | 74 ++++++++++++++++++ .../test/shared/relay/relay_session_test.dart | 77 ++++++++++++++++++- 4 files changed, 227 insertions(+), 10 deletions(-) create mode 100644 mobile/test/shared/relay/relay_query_signer_test.dart diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 70b175fad7..1126844e6e 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -76,6 +76,7 @@ class _BufferedEvent { class RelaySessionNotifier extends Notifier { 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 { client: httpClient, clientFactory: httpClientFactory, ), + _querySigner = querySigner ?? RelayQuerySigner(), _socketFactory = socketFactory, _now = now ?? DateTime.now, _rateLimitGate = rateLimitGate ?? RelayRateLimitGate(), @@ -93,6 +95,7 @@ class RelaySessionNotifier extends Notifier { _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 { 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 { void _dispose() { _disposed = true; _contextGeneration++; + _querySigner.retire(); _beforePauseCallbacks.clear(); _connectionGeneration++; _reconnectTimer?.cancel(); diff --git a/mobile/lib/shared/relay/relay_session_auth.dart b/mobile/lib/shared/relay/relay_session_auth.dart index 9c6c79ae02..042e54d842 100644 --- a/mobile/lib/shared/relay/relay_session_auth.dart +++ b/mobile/lib/shared/relay/relay_session_auth.dart @@ -1,5 +1,71 @@ part of 'relay_session.dart'; +/// Inputs passed to a query-auth signing worker. +typedef RelayQueryAuthRequest = ({ + String method, + String url, + List 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 Function(RelayQueryAuthRequest)? sign, + }) : _sign = sign ?? _signQueryInBackground; + + final Future Function(RelayQueryAuthRequest) _sign; + static const _maxPending = 64; + Future _tail = Future.value(); + int _pending = 0; + int _generation = 0; + + /// Signs one request, rejecting excess work without retaining another job. + Future sign(RelayQueryAuthRequest request) { + if (_pending >= _maxPending) { + return Future.error(StateError('Relay query signing queue is full')); + } + final generation = _generation; + _pending++; + final released = Completer(); + 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 _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, diff --git a/mobile/test/shared/relay/relay_query_signer_test.dart b/mobile/test/shared/relay/relay_query_signer_test.dart new file mode 100644 index 0000000000..2f7ac08615 --- /dev/null +++ b/mobile/test/shared/relay/relay_query_signer_test.dart @@ -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: [91, 93], + nsec: 'test-key', +); + +void main() { + test( + 'serializes workers and rejects saturation without retaining a job', + () async { + final worker = Completer(); + 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(); + 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'); + }); +} diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 3d57dca120..5f06e0b03c 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -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(); + final proof = Completer(); + 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(); 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()), + ); + await pumpEventQueue(); expect(clients.single.requestCount, 2); - await expectLater(timedOutQuery, throwsA(isA())); + 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.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); });