fix(discovery): handle udp datagram socket errors

lib/services/companion_remote/lan_discovery_service.dart

@@ -161,7 +161,7 @@ class LanDiscoveryService {
   Future<void> stopBroadcasting() async {
     _broadcastTimer?.cancel();
     _broadcastTimer = null;
-    _broadcastSockets?.close();
+    await _broadcastSockets?.close();
     _broadcastSockets = null;
     appLogger.d('LanDiscovery: Broadcasting stopped');
   }
@@ -222,14 +222,10 @@ class LanDiscoveryService {
 
       appLogger.d('LanDiscovery: Listening on port $discoveryPort');
 
-      _listenSubscription = _listenSocket!.listen((RawSocketEvent event) {
-        if (event == RawSocketEvent.read) {
-          final datagram = _listenSocket?.receive();
-          if (datagram != null) {
-            _handleDatagram(datagram, contexts);
-          }
-        }
-      });
+      _listenSubscription = _listenSocket!.listenDatagrams(
+        (datagram) => _handleDatagram(datagram, contexts),
+        debugLabel: 'LanDiscovery listener',
+      );
     } catch (e) {
       appLogger.e('LanDiscovery: Failed to bind listener', error: e);
     }

lib/services/jellyfin_lan_discovery_service.dart

@@ -25,23 +25,14 @@ class JellyfinLanDiscoveryService {
     InternetAddress? broadcastAddress,
   }) async {
     UdpBroadcastSocketSet? socketSet;
-    final subscriptions = <StreamSubscription<RawSocketEvent>>[];
     final discovered = <String, DiscoveredJellyfinServer>{};
     try {
       socketSet = await UdpBroadcastSockets.bind();
-      for (final socket in socketSet.sockets) {
-        subscriptions.add(
-          socket.listen((event) {
-            if (event != RawSocketEvent.read) return;
-            Datagram? datagram;
-            while ((datagram = socket.receive()) != null) {
-              final server = parseDiscoveryResponse(datagram!.data);
-              if (server == null) continue;
+      socketSet.listen((datagram) {
+        final server = parseDiscoveryResponse(datagram.data);
+        if (server == null) return;
         discovered.putIfAbsent(server.id, () => server);
-            }
-          }),
-        );
-      }
+      }, debugLabel: 'Jellyfin LAN discovery');
 
       final data = utf8.encode(discoveryMessage);
       final target = broadcastAddress ?? UdpBroadcastSockets.limitedBroadcastAddress;
@@ -52,10 +43,7 @@ class JellyfinLanDiscoveryService {
     } catch (e, st) {
       appLogger.w('Jellyfin LAN discovery failed', error: e, stackTrace: st);
     } finally {
-      for (final subscription in subscriptions) {
-        await subscription.cancel();
-      }
-      socketSet?.close();
+      await socketSet?.close();
     }
 
     return sortDiscoveredServers(discovered.values);

lib/utils/udp_broadcast_sockets.dart

@@ -5,13 +5,20 @@ import 'app_logger.dart';
 
 class UdpBroadcastSocketSet {
   final List<RawDatagramSocket> _sockets;
+  final List<StreamSubscription<RawSocketEvent>> _subscriptions = [];
 
-  const UdpBroadcastSocketSet._(this._sockets);
+  UdpBroadcastSocketSet._(this._sockets);
 
   bool get isEmpty => _sockets.isEmpty;
 
   Iterable<RawDatagramSocket> get sockets => _sockets;
 
+  void listen(void Function(Datagram datagram) onDatagram, {required String debugLabel}) {
+    for (final socket in _sockets) {
+      _subscriptions.add(socket.listenDatagrams(onDatagram, debugLabel: debugLabel));
+    }
+  }
+
   void send(List<int> data, InternetAddress address, int port) {
     for (final socket in _sockets) {
       try {
@@ -22,13 +29,42 @@ class UdpBroadcastSocketSet {
     }
   }
 
-  void close() {
+  Future<void> close() async {
+    final subscriptions = List<StreamSubscription<RawSocketEvent>>.of(_subscriptions);
+    _subscriptions.clear();
+    for (final subscription in subscriptions) {
+      try {
+        await subscription.cancel();
+      } catch (e, st) {
+        appLogger.w('UDP datagram subscription cancel failed', error: e, stackTrace: st);
+      }
+    }
     for (final socket in _sockets) {
       socket.close();
     }
   }
 }
 
+extension DatagramSocketListen on RawDatagramSocket {
+  StreamSubscription<RawSocketEvent> listenDatagrams(
+    void Function(Datagram datagram) onDatagram, {
+    required String debugLabel,
+  }) {
+    return listen(
+      (event) {
+        if (event != RawSocketEvent.read) return;
+        Datagram? datagram;
+        while ((datagram = receive()) != null) {
+          onDatagram(datagram!);
+        }
+      },
+      onError: (Object e, StackTrace st) {
+        appLogger.w('$debugLabel datagram socket error', error: e, stackTrace: st);
+      },
+    );
+  }
+}
+
 class UdpBroadcastSockets {
   UdpBroadcastSockets._();
 

test/services/jellyfin_lan_discovery_service_test.dart

@@ -1,7 +1,9 @@
 import 'dart:convert';
+import 'dart:io';
 
 import 'package:flutter_test/flutter_test.dart';
 import 'package:plezy/services/jellyfin_lan_discovery_service.dart';
+import 'package:plezy/utils/udp_broadcast_sockets.dart';
 
 void main() {
   group('JellyfinLanDiscoveryService', () {
@@ -41,5 +43,43 @@ void main() {
 
       expect(sorted.map((server) => server.id), ['srv-1', 'srv-2', 'srv-3']);
     });
+
+    test('listenDatagrams receives queued loopback datagrams', () async {
+      final receiver = await RawDatagramSocket.bind(InternetAddress.loopbackIPv4, 0);
+      final sender = await RawDatagramSocket.bind(InternetAddress.loopbackIPv4, 0);
+      final received = <String>[];
+      final subscription = receiver.listenDatagrams(
+        (datagram) => received.add(utf8.decode(datagram.data)),
+        debugLabel: 'JellyfinLanDiscoveryService test',
+      );
+
+      try {
+        sender.send(utf8.encode('one'), InternetAddress.loopbackIPv4, receiver.port);
+        sender.send(utf8.encode('two'), InternetAddress.loopbackIPv4, receiver.port);
+
+        await _waitFor(() => received.length >= 2);
+
+        expect(received, containsAll(['one', 'two']));
+      } finally {
+        await subscription.cancel();
+        receiver.close();
+        sender.close();
+      }
+    });
+
+    test('UdpBroadcastSocketSet close cancels owned datagram listeners', () async {
+      final socketSet = await UdpBroadcastSockets.bind();
+      socketSet.listen((_) {}, debugLabel: 'JellyfinLanDiscoveryService test');
+
+      await expectLater(socketSet.close(), completes);
+    });
   });
 }
+
+Future<void> _waitFor(bool Function() condition) async {
+  for (var i = 0; i < 50; i++) {
+    if (condition()) return;
+    await Future<void>.delayed(const Duration(milliseconds: 20));
+  }
+  fail('Timed out waiting for condition');
+}