fix(pubsub): remediate mid-stream errors, teardown races, and settings preservation

- Close StreamController on mid-stream gRPC errors in streamingPullWithStream.
- Set isCancelled synchronously before cancelAll in streamingPull teardown to prevent races.
- Preserve custom settings on Topic.create() and Subscription.create() by returning this.
- Align default maxMessages to 100 on PubSub.pull and update error docstrings.
- Pass totalTimeout: null in doc_examples.dart for unlimited streaming pull reconnection.
- Harmonize BatchingSettings error message to 'Must be greater than zero'.
- Add comprehensive tests for mid-stream error completion and 4-stream concurrency teardown.

TAG=agy
CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce

Author: sigurdm

pkgs/google_cloud_pubsub/CHANGELOG.md

@@ -1,5 +1,18 @@
 ## 0.1.0-wip
 
+- Preserved configured `publishSettings` and `ackSettings` when calling
+  `Topic.create()` and `Subscription.create()`, returning `this`.
+- Cleanly closed stream controller with `unawaited(controller.close())` on
+  mid-stream gRPC errors in `PubSub.streamingPullWithStream` to prevent
+  downstream consumers from hanging.
+- Aligned default `maxMessages` in `PubSub.pull` to 100 to match
+  `Subscription.pull`.
+- Synchronously set `isCancelled = true` before stream cancellation in
+  `Subscription.streamingPull` teardown on non-retryable errors or exhausted
+  retries to prevent duplicate errors and orphaned reconnect timers.
+- Harmonized error messages in `BatchingSettings` to 'Must be greater than
+  zero'.
+
 - Closed stream controller on setup failure in `PubSub.streamingPullWithStream`
   to prevent consumers from hanging.
 - Added immediate no-op on empty collections for `PubSub.publishMessages`,
@@ -31,9 +44,10 @@
 - Aligned error classification so `StatusCode.aborted` is retryable across
   raw `GrpcError` and mapped exceptions, while `StatusCode.dataLoss` is
   non-retryable across both.
-- Defaulted `totalTimeout` in custom `RetrySettings` passed to
-  `streamingPull` to `null` (indefinite reconnection) unless explicitly
-  specified.
+- Clarified reconnection timeout in `Subscription.streamingPull`: omitting retry
+  defaults to unlimited reconnection timeout (`totalTimeout: null`), while
+  custom `RetrySettings` retain their configured `totalTimeout` (default 1
+  minute) unless explicitly overridden.
 - Replaced constructor assertions in `BatchingSettings` and `RetrySettings`
   with always-on parameter validation, and updated `PublishSettings` and
   `AckSettings` constructors to default to newly created instances.

pkgs/google_cloud_pubsub/example/doc_examples.dart

@@ -26,6 +26,7 @@ Future<void> streamingPullExample(PubSub pubsub) async {
     streamAckDeadlineSeconds: 30,
     retry: RetrySettings(
       maxRetries: 10,
+      totalTimeout: null,
       initialDelay: const Duration(seconds: 1),
       maxDelay: const Duration(seconds: 30),
     ),

pkgs/google_cloud_pubsub/lib/src/batching.dart

@@ -42,11 +42,15 @@ final class BatchingSettings {
       throw ArgumentError.value(
         maxMessages,
         'maxMessages',
-        'Must be greater than 0',
+        'Must be greater than zero',
       );
     }
     if (maxBytes <= 0) {
-      throw ArgumentError.value(maxBytes, 'maxBytes', 'Must be greater than 0');
+      throw ArgumentError.value(
+        maxBytes,
+        'maxBytes',
+        'Must be greater than zero',
+      );
     }
     if (maxDelay <= Duration.zero) {
       throw ArgumentError.value(

pkgs/google_cloud_pubsub/lib/src/client.dart

@@ -361,12 +361,14 @@ final class PubSub {
   /// The [subscription] must be in the format
   /// `projects/<project-id>/subscriptions/<subscription-id>`.
   ///
+  /// It is an error if [maxMessages] is not greater than 0.
+  ///
   /// Throws a [NotFoundException] if the subscription does not exist.
   ///
   /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.Pull).
   Future<List<ReceivedMessage>> pull(
     String subscription, {
-    int maxMessages = 1,
+    int maxMessages = 100,
   }) async {
     if (maxMessages <= 0) {
       throw ArgumentError.value(
@@ -455,6 +457,7 @@ final class PubSub {
               } else {
                 controller.addError(e, s);
               }
+              unawaited(controller.close());
             },
             onDone: () {
               controller.close();

pkgs/google_cloud_pubsub/lib/src/subscription.dart

@@ -271,8 +271,10 @@ final class Subscription {
   /// Returns a [Subscription] instance representing the created subscription.
   ///
   /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.CreateSubscription).
-  Future<Subscription> create({required String topic}) =>
-      pubsub.createSubscription(name, topic: topic);
+  Future<Subscription> create({required String topic}) async {
+    await pubsub.createSubscription(name, topic: topic);
+    return this;
+  }
 
   /// Deletes this subscription on the server.
   ///
@@ -283,6 +285,9 @@ final class Subscription {
 
   /// Pulls up to [maxMessages] from this subscription.
   ///
+  /// It is an error if [maxMessages] is not greater than 0.
+  /// It is an error if called on a closed [Subscription].
+  ///
   /// Throws a [NotFoundException] if the subscription does not exist.
   ///
   /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Subscriber.Pull).
@@ -311,7 +316,9 @@ final class Subscription {
   ///
   /// The stream automatically reconnects on transient network errors using the
   /// configured [retry] settings (defaulting to [AckSettings.retry] with
-  /// unlimited total duration).
+  /// unlimited total duration). Custom [RetrySettings] retain their configured
+  /// [RetrySettings.totalTimeout] (which defaults to 1 minute) unless
+  /// `totalTimeout: null` is passed for unlimited reconnection duration.
   /// Reconnections use exponential backoff, which resets once a connection
   /// has been sustained and healthy (>= 15 seconds) or successfully yields
   /// messages.
@@ -454,6 +461,7 @@ final class Subscription {
         if (_isClosed || isCancelled || controller.isClosed) return;
 
         if (error != null && !isRetryable(error)) {
+          isCancelled = true;
           controller.addError(error, stackTrace);
           await cancelAll();
           activeOrReconnectingStreams = 0;
@@ -489,6 +497,7 @@ final class Subscription {
             controller.addError(error, stackTrace);
           }
           if (activeOrReconnectingStreams == 0) {
+            isCancelled = true;
             await cancelAll();
             _activeStreamingPullControllers.remove(controller);
             await controller.close();

pkgs/google_cloud_pubsub/lib/src/topic.dart

@@ -159,7 +159,10 @@ final class Topic {
   /// Returns a [Topic] instance representing the created topic.
   ///
   /// See the [official documentation](https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.Publisher.CreateTopic).
-  Future<Topic> create() => pubsub.createTopic(name);
+  Future<Topic> create() async {
+    await pubsub.createTopic(name);
+    return this;
+  }
 
   /// Deletes this topic on the server.
   ///

pkgs/google_cloud_pubsub/test/batching_test.dart

@@ -187,5 +187,78 @@ void main() {
         throwsA(isA<ArgumentError>()),
       );
     });
+
+    test('parameter validation error messages are harmonized', () {
+      expect(
+        () => BatchingSettings(maxMessages: 0),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+      expect(
+        () => BatchingSettings(maxMessages: -1),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+      expect(
+        () => BatchingSettings(maxBytes: 0),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+      expect(
+        () => BatchingSettings(maxBytes: -10),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+      expect(
+        () => BatchingSettings(maxDelay: Duration.zero),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+      expect(
+        () => BatchingSettings(maxDelay: const Duration(milliseconds: -1)),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+      expect(
+        () => BatchingSettings(maxDelay: const Duration(seconds: -10)),
+        throwsA(
+          isA<ArgumentError>().having(
+            (e) => e.message,
+            'message',
+            'Must be greater than zero',
+          ),
+        ),
+      );
+    });
   });
 }

pkgs/google_cloud_pubsub/test/lifecycle_test.dart

@@ -85,6 +85,12 @@ class FakePublisherClient extends Fake implements generated.PublisherClient {
     }
     return FakeResponseFuture(completer.future);
   }
+
+  @override
+  grpc.ResponseFuture<generated.Topic> createTopic(
+    generated.Topic request, {
+    grpc.CallOptions? options,
+  }) => FakeResponseFuture(Future.value(request));
 }
 
 class FakeSubscriberClient extends Fake implements generated.SubscriberClient {
@@ -137,6 +143,12 @@ class FakeSubscriberClient extends Fake implements generated.SubscriberClient {
     lastMaxMessages = request.maxMessages;
     return FakeResponseFuture(Future.value(generated.PullResponse()));
   }
+
+  @override
+  grpc.ResponseFuture<generated.Subscription> createSubscription(
+    generated.Subscription request, {
+    grpc.CallOptions? options,
+  }) => FakeResponseFuture(Future.value(request));
 }
 
 class FakeClientChannel extends Fake implements grpc.ClientChannel {
@@ -303,6 +315,17 @@ void main() {
         expect(fakePublisher.publishCalled, isFalse);
       },
     );
+
+    test('Topic.create() returns this and preserves publishSettings', () async {
+      final customSettings = PublishSettings(
+        batching: BatchingSettings(maxMessages: 42),
+      );
+      final topic = client.topic('my-topic', publishSettings: customSettings);
+      final created = await topic.create();
+      expect(identical(created, topic), isTrue);
+      expect(created.publishSettings, same(customSettings));
+      expect(created.publishSettings.batching.maxMessages, equals(42));
+    });
   });
 
   group('Subscription Lifecycle & Batcher', () {
@@ -412,6 +435,22 @@ void main() {
         expect(fakeSubscriber.modifyAckDeadlineCalled, isFalse);
       },
     );
+
+    test(
+      'Subscription.create() returns this and preserves ackSettings',
+      () async {
+        final customSettings = AckSettings(
+          batching: BatchingSettings(maxMessages: 42),
+        );
+        final sub = client.subscription('my-sub', ackSettings: customSettings);
+        final created = await sub.create(
+          topic: 'projects/test-project/topics/my-topic',
+        );
+        expect(identical(created, sub), isTrue);
+        expect(created.ackSettings, same(customSettings));
+        expect(created.ackSettings.batching.maxMessages, equals(42));
+      },
+    );
   });
 
   group('PubSub Client Empty List No-Ops & Parameter Validation', () {
@@ -523,5 +562,18 @@ void main() {
         throwsA(isA<ArgumentError>()),
       );
     });
+
+    test('PubSub.pull defaults to maxMessages 100', () async {
+      await client.pull('projects/test-project/subscriptions/my-sub');
+      expect(fakeSubscriber.pullCalled, isTrue);
+      expect(fakeSubscriber.lastMaxMessages, equals(100));
+    });
+
+    test('Subscription.pull defaults to maxMessages 100', () async {
+      final sub = client.subscription('my-sub');
+      await sub.pull();
+      expect(fakeSubscriber.pullCalled, isTrue);
+      expect(fakeSubscriber.lastMaxMessages, equals(100));
+    });
   });
 }