feat(pubsub): Add exponential backoff, batching, and parallel streaming pull - #292
feat(pubsub): Add exponential backoff, batching, and parallel streaming pull#292sigurdm wants to merge 19 commits into
Conversation
…ng pull - Implemented exponential backoff for retries. - Added background batching for acknowledge and modifyAckDeadline. - Re-wrote streamingPull to use StreamController instead of async* to prevent cancellation hanging. - Added maxConcurrentStreams for high-throughput parallel streams. - Fixed stream leak/state errors in streaming reconnects. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new experimental package google_cloud_pubsub for Google Cloud Pub/Sub in Dart, and updates google_cloud_logging and google_cloud_shelf to support request-log correlation using Dart Zones and W3C traceparent headers. Feedback on the new Pub/Sub client identifies potential infinite loops/busy-waiting in _onAckBatch and _onModifyAckBatch when active streams are empty, suggesting a fallback to unary RPCs to prevent high CPU usage and unacknowledged messages.
| while (_streamingPullSubscriptionCount > 0) { | ||
| if (_activeStreams.isEmpty) { | ||
| await Future<void>.delayed(const Duration(milliseconds: 100)); | ||
| continue; | ||
| } | ||
|
|
||
| // Simple round-robin load balancing. | ||
| _nextStreamIndex = (_nextStreamIndex + 1) % _activeStreams.length; | ||
| final stream = _activeStreams[_nextStreamIndex]; | ||
|
|
||
| if (stream.isClosed) { | ||
| await Future<void>.delayed(const Duration(milliseconds: 10)); | ||
| continue; | ||
| } | ||
|
|
||
| try { | ||
| stream.add(grpc.StreamingPullRequest()..ackIds.addAll(ackIds)); | ||
| return; | ||
| } on StateError { | ||
| // ignore: avoid_catching_errors | ||
| // Stream was closed concurrently after the isClosed check. | ||
| continue; | ||
| } | ||
| } |
There was a problem hiding this comment.
The current implementation of _onAckBatch can enter an infinite loop of Future.delayed if _streamingPullSubscriptionCount > 0 but _activeStreams is empty (e.g., during stream reconnection or after all streams have failed). This can lead to high CPU usage, memory leaks, and unacknowledged messages that eventually expire and get redelivered.
Instead of busy-waiting indefinitely, we should limit the number of attempts to the number of available streams. If no active, open stream can be found, we should break out of the loop and fall back to the unary pubsub.acknowledge RPC. This guarantees that acknowledgments are delivered promptly even when the streaming connection is temporarily down.
var attempts = 0;
while (_streamingPullSubscriptionCount > 0 &&
_activeStreams.isNotEmpty &&
attempts < _activeStreams.length) {
attempts++;
_nextStreamIndex = (_nextStreamIndex + 1) % _activeStreams.length;
final stream = _activeStreams[_nextStreamIndex];
if (stream.isClosed) {
continue;
}
try {
stream.add(grpc.StreamingPullRequest()..ackIds.addAll(ackIds));
return;
} on StateError {
// ignore: avoid_catching_errors
// Stream was closed concurrently after the isClosed check.
continue;
}
}| while (_streamingPullSubscriptionCount > 0) { | ||
| if (_activeStreams.isEmpty) { | ||
| await Future<void>.delayed(const Duration(milliseconds: 100)); | ||
| continue; | ||
| } | ||
|
|
||
| // Simple round-robin load balancing. | ||
| _nextStreamIndex = (_nextStreamIndex + 1) % _activeStreams.length; | ||
| final stream = _activeStreams[_nextStreamIndex]; | ||
|
|
||
| if (stream.isClosed) { | ||
| await Future<void>.delayed(const Duration(milliseconds: 10)); | ||
| continue; | ||
| } | ||
|
|
||
| try { | ||
| stream.add( | ||
| grpc.StreamingPullRequest() | ||
| ..modifyDeadlineAckIds.addAll(ackIds) | ||
| ..modifyDeadlineSeconds.addAll( | ||
| List.filled(ackIds.length, deadline), | ||
| ), | ||
| ); | ||
| return; | ||
| } on StateError { | ||
| // ignore: avoid_catching_errors | ||
| // Stream was closed concurrently after the isClosed check. | ||
| continue; | ||
| } | ||
| } |
There was a problem hiding this comment.
The current implementation of _onModifyAckBatch can enter an infinite loop of Future.delayed if _streamingPullSubscriptionCount > 0 but _activeStreams is empty (e.g., during stream reconnection or after all streams have failed). This can lead to high CPU usage, memory leaks, and unacknowledged messages that eventually expire and get redelivered.
Instead of busy-waiting indefinitely, we should limit the number of attempts to the number of available streams. If no active, open stream can be found, we should break out of the loop and fall back to the unary pubsub.modifyAckDeadline RPC. This guarantees that ack deadline modifications are delivered promptly even when the streaming connection is temporarily down.
var attempts = 0;
while (_streamingPullSubscriptionCount > 0 &&
_activeStreams.isNotEmpty &&
attempts < _activeStreams.length) {
attempts++;
_nextStreamIndex = (_nextStreamIndex + 1) % _activeStreams.length;
final stream = _activeStreams[_nextStreamIndex];
if (stream.isClosed) {
continue;
}
try {
stream.add(
grpc.StreamingPullRequest()
..modifyDeadlineAckIds.addAll(ackIds)
..modifyDeadlineSeconds.addAll(
List.filled(ackIds.length, deadline),
),
);
return;
} on StateError {
// ignore: avoid_catching_errors
// Stream was closed concurrently after the isClosed check.
continue;
}
}# Conflicts: # pkgs/google_cloud_pubsub/lib/src/subscription.dart # pkgs/google_cloud_pubsub/test/integration_test.dart
- Replaced the busy-waiting loop in _onAckBatch and _onModifyAckBatch with a single-pass check over currently active streams. - If no active streams are available or they all fail concurrently, immediately fall back to the unary RPC. - This prevents high CPU usage and ensures ACKs are not blocked indefinitely when streams are temporarily down. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
- Removed _streamingPullSubscriptionCount field and simplified active stream checks by relying solely on _activeStreams.isNotEmpty. - Fixed relative import and directives ordering in subscription.dart. - Added ignore_for_file for avoid_catching_errors to suppress necessary StateError catches cleanly. - Fixed expression body lint in client.dart. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
…and resolve lints
- Refactored _onModifyAckBatch to use Dart 3 MapEntry pattern destructuring.
- Added extensive, high-quality dartdocs with preconditions, exception lists, and performance considerations for Subscription, AckSettings, and retry utilities.
- Implemented doc_examples.dart to serve as external sources for {@example} references in dartdoc.
- Resolved all static analysis warnings (unused variables, duplicate ignores, and incorrect comment references).
TAG=agy
CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
- Replaced vague GrpcError wrapper references with specific and actionable PubSubOperationException in acknowledgeNow and modifyAckDeadlineNow docs. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
- Changed Throws [ArgumentError] references to It is an error if... to match standard project style for programmer errors. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
|
Where are we here? |
2 participants
This PR is stacked on top of #244.
It implements the following enhancements:
acknowledgeNowandmodifyAckDeadlineNownow support background batching for performance.maxConcurrentStreamstostreamingPullto bypass single gRPC stream limits.async*broadcast streams, fixed duplicate reconnects, and preventedStateErrorcrashes.