fix(mobile): bound one-shot relay attempts and release failed auth

Signed-off-by: Tom Brow <tomb@block.xyz>

mobile/lib/shared/relay/relay_socket.dart

@@ -39,6 +39,7 @@ class RelaySocket {
 
   final String _wsUrl;
   final String? _nsec;
+  final Duration? _connectTimeout;
   final void Function(List<dynamic> message) _onMessage;
   final void Function() _onConnected;
   final void Function(Object? error) _onDisconnected;
@@ -58,8 +59,10 @@ class RelaySocket {
     required void Function(List<dynamic> message) onMessage,
     required void Function() onConnected,
     required void Function(Object? error) onDisconnected,
+    Duration? connectTimeout,
   }) : _wsUrl = wsUrl,
        _nsec = nsec,
+       _connectTimeout = connectTimeout,
        _onMessage = onMessage,
        _onConnected = onConnected,
        _onDisconnected = onDisconnected;
@@ -73,9 +76,13 @@ class RelaySocket {
       _channel = IOWebSocketChannel.connect(
         Uri.parse(_wsUrl),
         pingInterval: debugPingInterval,
+        connectTimeout: _connectTimeout,
       );
       await _channel!.ready;
     } catch (e) {
+      // A failed handshake also queues the error on the channel stream. No
+      // normal subscription exists yet; cancel it so sink.close can complete.
+      await _channel?.stream.listen((_) {}, onError: (Object _) {}).cancel();
       _state = SocketState.disconnected;
       _onDisconnected(e);
       return;
@@ -133,6 +140,7 @@ class RelaySocket {
 
   /// Gracefully close the connection.
   Future<void> disconnect() async {
+    _failAuth(Exception('Relay disconnected during authentication'));
     _resetConnection();
     final channel = _channel;
     _channel = null;

mobile/lib/shared/relay/signed_event_relay.dart

@@ -64,7 +64,8 @@ class SignedEventRelay {
 /// Publishes one signed event over a short-lived authenticated NIP-42 socket.
 ///
 /// This is used for community-removal tombstones because the community being
-/// removed is not necessarily the app's active relay session.
+/// removed is not necessarily the app's active relay session. The deadline
+/// covers connection, authentication, and acknowledgement.
 Future<NostrEvent> submitSignedEventOnce({
   required String wsUrl,
   required String nsec,
@@ -73,6 +74,7 @@ Future<NostrEvent> submitSignedEventOnce({
   required List<List<String>> tags,
   int? createdAt,
   Duration timeout = const Duration(seconds: 12),
+  RelaySocketFactory socketFactory = _oneShotSocket,
 }) async {
   final privateKey = nostr.Nip19.decode(payload: nsec).data;
   if (privateKey.isEmpty) throw const FormatException('Invalid nsec');
@@ -87,7 +89,7 @@ Future<NostrEvent> submitSignedEventOnce({
   final event = NostrEvent.fromJson(signed.toMap());
   final result = Completer<NostrEvent>();
   late final RelaySocket socket;
-  socket = RelaySocket(
+  socket = socketFactory(
     wsUrl: wsUrl,
     nsec: nsec,
     onMessage: (message) {
@@ -122,11 +124,28 @@ Future<NostrEvent> submitSignedEventOnce({
       }
     },
   );
-  final resultFuture = result.future.timeout(timeout);
   try {
-    await socket.connect();
-    return await resultFuture;
+    await Future.wait<void>([
+      socket.connect(),
+      result.future.then<void>((_) {}),
+    ], eagerError: true).timeout(timeout);
+    return await result.future;
   } finally {
     await socket.disconnect();
   }
 }
+
+RelaySocket _oneShotSocket({
+  required String wsUrl,
+  required String? nsec,
+  required void Function(List<dynamic>) onMessage,
+  required void Function() onConnected,
+  required void Function(Object? error) onDisconnected,
+}) => RelaySocket(
+  wsUrl: wsUrl,
+  nsec: nsec,
+  onMessage: onMessage,
+  onConnected: onConnected,
+  onDisconnected: onDisconnected,
+  connectTimeout: const Duration(seconds: 8),
+);

mobile/test/shared/relay/relay_socket_liveness_test.dart

@@ -5,6 +5,7 @@ import 'dart:io';
 import 'package:buzz/shared/relay/relay_socket.dart';
 import 'package:crypto/crypto.dart';
 import 'package:flutter_test/flutter_test.dart';
+import 'package:nostr/nostr.dart' as nostr;
 
 /// A server that completes the WS handshake then never speaks again: no pongs,
 /// no close frame. Only a client-side ping timeout can notice.
@@ -51,6 +52,43 @@ void main() {
     RelaySocket.debugPingInterval = RelaySocket.pingInterval;
   });
 
+  test('disconnect completes pending authentication', () async {
+    RelaySocket.debugPingInterval = RelaySocket.pingInterval;
+    final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
+    final authReceived = Completer<void>();
+    WebSocket? peer;
+    server.listen((request) async {
+      final connection = await WebSocketTransformer.upgrade(request);
+      peer = connection;
+      connection.listen((raw) {
+        final frame = jsonDecode(raw as String) as List<dynamic>;
+        if (frame.first == 'AUTH' && !authReceived.isCompleted) {
+          authReceived.complete();
+        }
+      });
+      connection.add(jsonEncode(['AUTH', 'fixture-challenge']));
+    });
+    final socket = RelaySocket(
+      wsUrl: 'ws://127.0.0.1:${server.port}',
+      nsec: nostr.Keys.generate().nsec,
+      onMessage: (_) {},
+      onConnected: () => fail('The peer never acknowledges authentication'),
+      onDisconnected: (_) {},
+    );
+    final connecting = socket.connect();
+    try {
+      await authReceived.future;
+      expect(socket.state, SocketState.authenticating);
+      await socket.disconnect();
+      await connecting.timeout(const Duration(seconds: 1));
+      expect(socket.state, SocketState.disconnected);
+    } finally {
+      socket.dispose();
+      await peer?.close();
+      await server.close(force: true);
+    }
+  });
+
   test('detects a peer that stops answering pings', () async {
     final server = await _silentAfterHandshakeServer();
 

mobile/test/shared/relay/signed_event_relay_test.dart

@@ -0,0 +1,212 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:buzz/shared/relay/nostr_models.dart';
+import 'package:buzz/shared/relay/relay_socket.dart';
+import 'package:buzz/shared/relay/signed_event_relay.dart';
+import 'package:fake_async/fake_async.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:nostr/nostr.dart' as nostr;
+
+class _ControlledSocket extends RelaySocket {
+  final Completer<void>? connectionGate;
+  final void Function() connected;
+  final void Function(List<dynamic>) message;
+  bool disconnected = false;
+  bool eventSent = false;
+  bool acknowledge = false;
+
+  _ControlledSocket({
+    required super.wsUrl,
+    required super.nsec,
+    required super.onMessage,
+    required super.onConnected,
+    required super.onDisconnected,
+    this.connectionGate,
+  }) : connected = onConnected,
+       message = onMessage;
+
+  @override
+  Future<void> connect() async {
+    if (connectionGate != null) await connectionGate!.future;
+    if (!disconnected) connected();
+  }
+
+  @override
+  void send(List<dynamic> payload) {
+    eventSent = true;
+    if (acknowledge) {
+      final event = payload[1] as Map<String, dynamic>;
+      message(['OK', event['id'], true, 'saved']);
+    }
+  }
+
+  @override
+  Future<void> disconnect() async {
+    disconnected = true;
+    if (connectionGate != null && !connectionGate!.isCompleted) {
+      connectionGate!.complete();
+    }
+  }
+}
+
+void main() {
+  final key = nostr.Keys.generate();
+  const timeout = Duration(seconds: 12);
+
+  test('connection timeout closes a stalled WebSocket handshake', () async {
+    final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
+    server.listen((request) {});
+    try {
+      await expectLater(
+        submitSignedEventOnce(
+          wsUrl: 'ws://127.0.0.1:${server.port}',
+          nsec: key.nsec,
+          kind: 30350,
+          content: 'inactive fixture',
+          tags: const [],
+          socketFactory:
+              ({
+                required wsUrl,
+                required nsec,
+                required onMessage,
+                required onConnected,
+                required onDisconnected,
+              }) => RelaySocket(
+                wsUrl: wsUrl,
+                nsec: nsec,
+                onMessage: onMessage,
+                onConnected: onConnected,
+                onDisconnected: onDisconnected,
+                connectTimeout: const Duration(milliseconds: 50),
+              ),
+        ).timeout(const Duration(seconds: 2)),
+        throwsA(
+          isA<TimeoutException>().having(
+            (error) => error.duration,
+            'connection deadline',
+            const Duration(milliseconds: 50),
+          ),
+        ),
+      );
+    } finally {
+      await server.close(force: true);
+    }
+  });
+
+  test('one-shot deadline includes a stalled connection', () {
+    fakeAsync((clock) {
+      final gate = Completer<void>();
+      late _ControlledSocket socket;
+      final errors = <Object>[];
+      final publication = submitSignedEventOnce(
+        wsUrl: 'wss://relay.example',
+        nsec: key.nsec,
+        kind: 30350,
+        content: 'inactive fixture',
+        tags: const [],
+        timeout: timeout,
+        socketFactory:
+            ({
+              required wsUrl,
+              required nsec,
+              required onMessage,
+              required onConnected,
+              required onDisconnected,
+            }) => socket = _ControlledSocket(
+              wsUrl: wsUrl,
+              nsec: nsec,
+              onMessage: onMessage,
+              onConnected: onConnected,
+              onDisconnected: onDisconnected,
+              connectionGate: gate,
+            ),
+      );
+      unawaited(
+        publication.then<void>((_) {}, onError: (Object e) => errors.add(e)),
+      );
+      clock.flushMicrotasks();
+      try {
+        clock.elapse(timeout);
+        clock.flushMicrotasks();
+        expect(errors, [isA<TimeoutException>()]);
+        expect(socket.disconnected, isTrue);
+        expect(socket.eventSent, isFalse);
+      } finally {
+        if (!gate.isCompleted) gate.complete();
+        clock.flushMicrotasks();
+      }
+    });
+  });
+
+  test('one-shot deadline bounds waiting for relay acknowledgement', () {
+    fakeAsync((clock) {
+      late _ControlledSocket socket;
+      final errors = <Object>[];
+      unawaited(
+        submitSignedEventOnce(
+          wsUrl: 'wss://relay.example',
+          nsec: key.nsec,
+          kind: 30350,
+          content: 'inactive fixture',
+          tags: const [],
+          socketFactory:
+              ({
+                required wsUrl,
+                required nsec,
+                required onMessage,
+                required onConnected,
+                required onDisconnected,
+              }) => socket = _ControlledSocket(
+                wsUrl: wsUrl,
+                nsec: nsec,
+                onMessage: onMessage,
+                onConnected: onConnected,
+                onDisconnected: onDisconnected,
+              ),
+        ).then<void>((_) {}, onError: (Object e) => errors.add(e)),
+      );
+      clock.flushMicrotasks();
+      expect(socket.eventSent, isTrue);
+      clock.elapse(timeout);
+      expect(errors, [isA<TimeoutException>()]);
+      expect(socket.disconnected, isTrue);
+    });
+  });
+
+  test('accepted one-shot publication closes its temporary socket', () {
+    fakeAsync((clock) {
+      late _ControlledSocket socket;
+      final accepted = <NostrEvent>[];
+      unawaited(
+        submitSignedEventOnce(
+          wsUrl: 'wss://relay.example',
+          nsec: key.nsec,
+          kind: 30350,
+          content: 'inactive fixture',
+          tags: const [],
+          socketFactory:
+              ({
+                required wsUrl,
+                required nsec,
+                required onMessage,
+                required onConnected,
+                required onDisconnected,
+              }) {
+                socket = _ControlledSocket(
+                  wsUrl: wsUrl,
+                  nsec: nsec,
+                  onMessage: onMessage,
+                  onConnected: onConnected,
+                  onDisconnected: onDisconnected,
+                )..acknowledge = true;
+                return socket;
+              },
+        ).then(accepted.add),
+      );
+      clock.flushMicrotasks();
+      expect(accepted.single.content, 'saved');
+      expect(socket.disconnected, isTrue);
+    });
+  });
+}