fix(query_manager): guard late stream events after request timeout

Guard the response stream's onData handler the same way onError is
guarded, and cancel the subscription once the completer settles.

Stream.timeout forwards a TimeoutException without cancelling the source,
so a successful response arriving after the timeout reached an unguarded
completer.complete on an already-completed completer, throwing
"Bad state: Future already completed" from an async callback that escaped
the surrounding try/catch and crashed the app. Also complete with an
error on an empty stream so the future cannot hang, restoring the prior
responseStream.first semantics.

Fixes #1525

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: vincenzopalazzo

packages/graphql/lib/src/core/query_manager.dart

@@ -281,14 +281,42 @@ class QueryManager {
         responseStream = responseStream.timeout(timeout);
       }
 
-      // Listen for the first response or error
-      responseStream.listen(completer.complete,
-          onError: (Object error, StackTrace stackTrace) {
-        if (!completer.isCompleted) {
-          // We return the first error encountered
-          completer.completeError(error, stackTrace);
-        }
-      });
+      // Listen for the first response or error.
+      //
+      // Both `onData` and `onError` must be guarded against completing the
+      // completer more than once. `Stream.timeout` forwards a
+      // `TimeoutException` to `onError` without cancelling the source
+      // subscription, so a late event (e.g. a successful response that
+      // arrives after the timeout has already settled the completer) can
+      // still be delivered. Completing an already-completed completer throws
+      // "Bad state: Future already completed" from inside this async callback,
+      // which escapes the surrounding try/catch and crashes the app.
+      //
+      // Once the completer is settled we cancel the subscription so the
+      // underlying request does not keep running after we have a result.
+      late final StreamSubscription<Response> subscription;
+      subscription = responseStream.listen(
+        (response) {
+          if (!completer.isCompleted) {
+            completer.complete(response);
+          }
+          subscription.cancel();
+        },
+        onError: (Object error, StackTrace stackTrace) {
+          if (!completer.isCompleted) {
+            // We return the first error encountered
+            completer.completeError(error, stackTrace);
+          }
+          subscription.cancel();
+        },
+        onDone: () {
+          // The stream closed without emitting; surface this the same way
+          // `Stream.first` used to so the awaiting future does not hang.
+          if (!completer.isCompleted) {
+            completer.completeError(StateError('No element'));
+          }
+        },
+      );
 
       // Await the response or error
       response = await completer.future;

packages/graphql/test/query_manager_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:gql/language.dart';
 import 'package:graphql/client.dart';
 import 'package:mockito/mockito.dart';
@@ -42,5 +44,57 @@ void main() {
       );
       client.queryManager.refetchQuery<dynamic>(observable.queryId);
     });
+
+    // Regression test for https://github.com/zino-hofmann/graphql-flutter/issues/1525
+    //
+    // When a request times out and the underlying link later delivers a
+    // successful response, the late event must be ignored rather than
+    // completing the already-settled completer (which used to throw an
+    // uncaught "Bad state: Future already completed").
+    test(
+      "late response after queryRequestTimeout does not crash",
+      () async {
+        final controller = StreamController<Response>();
+        when(
+          link.request(any),
+        ).thenAnswer((_) => controller.stream);
+
+        final timeoutClient = GraphQLClient(
+          cache: getTestCache(),
+          link: link,
+          queryRequestTimeout: const Duration(milliseconds: 50),
+        );
+
+        // The request never emits before the timeout fires.
+        final result = await timeoutClient.query(
+          QueryOptions<dynamic>(
+            document: parseString("{ fetchPerson { name } }"),
+          ),
+        );
+
+        expect(result.hasException, isTrue);
+        expect(result.exception!.linkException, isA<UnknownException>());
+        expect(
+          (result.exception!.linkException as UnknownException)
+              .originalException,
+          isA<TimeoutException>(),
+        );
+
+        // The real response arrives after the timeout already settled the
+        // request. This must not throw an uncaught error.
+        controller.add(
+          Response(
+            data: <String, dynamic>{
+              'fetchPerson': <String, dynamic>{'name': 'late'},
+            },
+            response: <String, dynamic>{},
+          ),
+        );
+        await controller.close();
+
+        // Pump the event loop so any late delivery would surface here.
+        await Future<void>.delayed(const Duration(milliseconds: 20));
+      },
+    );
   });
 }