Skip to content

Commit 5ad97bc

Browse files
sigurdmsigurdm
committed
feat(pubsub): add exponential backoff retry infrastructure
Add exponential backoff retry settings (RetrySettings) with jitter, totalTimeout support, and transient error classification (isRetryable) to package:google_cloud_pubsub. TAG=agy CONV=bc7c6164-e9cb-4a0f-b326-7752c814e1ce
1 parent f2e1c1c commit 5ad97bc

5 files changed

Lines changed: 792 additions & 0 deletions

File tree

‎pkgs/google_cloud_pubsub/CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
## 0.1.0-wip
22

3+
- Added `RetrySettings` to configure exponential backoff retry parameters
4+
(delays, multiplier, jitter, and total timeout).
35
- Initial release of the experimental Google Cloud Pub/Sub client.
46
- Supports basic topic and subscription management.
57
- Supports publishing and pulling messages (including streaming pull).

‎pkgs/google_cloud_pubsub/lib/google_cloud_pubsub.dart‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,5 +22,6 @@ export 'package:grpc/grpc.dart'
2222

2323
export 'src/client.dart' show PubSub;
2424
export 'src/message.dart' show Message, ReceivedMessage;
25+
export 'src/retry.dart' show RetrySettings;
2526
export 'src/subscription.dart' show Subscription;
2627
export 'src/topic.dart' show Topic;

‎pkgs/google_cloud_pubsub/lib/src/retry.dart‎

Lines changed: 251 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,251 @@
1+
// Copyright 2026 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
import 'dart:math';
16+
17+
import 'package:clock/clock.dart';
18+
import 'package:google_cloud_rpc/exceptions.dart';
19+
import 'package:grpc/grpc.dart';
20+
import 'package:meta/meta.dart';
21+
22+
const _sentinel = Object();
23+
24+
/// Settings for configuring retry logic with exponential backoff.
25+
final class RetrySettings {
26+
/// The maximum number of times to retry before failing.
27+
///
28+
/// A `null` value indicates that the number of retries is unlimited.
29+
final int? maxRetries;
30+
31+
/// The maximum amount of total time to retry before failing.
32+
///
33+
/// A `null` value indicates that the total retry time is unlimited.
34+
final Duration? totalTimeout;
35+
36+
/// The minimum amount of time to wait before retrying.
37+
final Duration initialDelay;
38+
39+
/// The multiplier for the wait time between retries.
40+
final double delayMultiplier;
41+
42+
/// The maximum amount of time to wait between retries.
43+
///
44+
/// If the calculated exponential wait time between retries exceeds this
45+
/// value, the wait time will be clamped to this value.
46+
final Duration maxDelay;
47+
48+
/// Creates a new [RetrySettings] instance.
49+
///
50+
/// It is an error if:
51+
/// - [maxRetries] is negative.
52+
/// - [initialDelay] is not greater than [Duration.zero].
53+
/// - [delayMultiplier] is less than 1.0 or not finite.
54+
/// - [maxDelay] is not greater than [Duration.zero].
55+
/// - [totalTimeout] is not greater than [Duration.zero].
56+
RetrySettings({
57+
int? maxRetries,
58+
Duration? totalTimeout = const Duration(minutes: 1),
59+
Duration initialDelay = const Duration(milliseconds: 100),
60+
double delayMultiplier = 1.3,
61+
Duration maxDelay = const Duration(seconds: 60),
62+
}) : this._internal(
63+
maxRetries: maxRetries,
64+
totalTimeout: totalTimeout,
65+
initialDelay: initialDelay,
66+
delayMultiplier: delayMultiplier,
67+
maxDelay: maxDelay,
68+
);
69+
70+
RetrySettings._internal({
71+
required this.maxRetries,
72+
required this.totalTimeout,
73+
required this.initialDelay,
74+
required this.delayMultiplier,
75+
required this.maxDelay,
76+
}) {
77+
if (maxRetries != null && maxRetries! < 0) {
78+
throw ArgumentError.value(
79+
maxRetries,
80+
'maxRetries',
81+
'Must be non-negative',
82+
);
83+
}
84+
if (initialDelay <= Duration.zero) {
85+
throw ArgumentError.value(
86+
initialDelay,
87+
'initialDelay',
88+
'Must be greater than zero',
89+
);
90+
}
91+
if (delayMultiplier < 1.0 || !delayMultiplier.isFinite) {
92+
throw ArgumentError.value(
93+
delayMultiplier,
94+
'delayMultiplier',
95+
'Must be at least 1.0',
96+
);
97+
}
98+
if (maxDelay <= Duration.zero) {
99+
throw ArgumentError.value(
100+
maxDelay,
101+
'maxDelay',
102+
'Must be greater than zero',
103+
);
104+
}
105+
if (totalTimeout != null && totalTimeout! <= Duration.zero) {
106+
throw ArgumentError.value(
107+
totalTimeout,
108+
'totalTimeout',
109+
'Must be greater than zero',
110+
);
111+
}
112+
}
113+
114+
/// Creates a copy of this [RetrySettings] with the given fields replaced.
115+
///
116+
/// It is an error if any replaced parameter violates its constraints.
117+
RetrySettings copyWith({
118+
Object? maxRetries = _sentinel,
119+
Object? totalTimeout = _sentinel,
120+
Duration? initialDelay,
121+
double? delayMultiplier,
122+
Duration? maxDelay,
123+
}) {
124+
if (maxRetries != _sentinel && maxRetries != null && maxRetries is! int) {
125+
throw ArgumentError.value(
126+
maxRetries,
127+
'maxRetries',
128+
'Must be an int or null',
129+
);
130+
}
131+
if (totalTimeout != _sentinel &&
132+
totalTimeout != null &&
133+
totalTimeout is! Duration) {
134+
throw ArgumentError.value(
135+
totalTimeout,
136+
'totalTimeout',
137+
'Must be a Duration or null',
138+
);
139+
}
140+
return RetrySettings._internal(
141+
maxRetries: identical(maxRetries, _sentinel)
142+
? this.maxRetries
143+
: maxRetries as int?,
144+
totalTimeout: identical(totalTimeout, _sentinel)
145+
? this.totalTimeout
146+
: totalTimeout as Duration?,
147+
initialDelay: initialDelay ?? this.initialDelay,
148+
delayMultiplier: delayMultiplier ?? this.delayMultiplier,
149+
maxDelay: maxDelay ?? this.maxDelay,
150+
);
151+
}
152+
}
153+
154+
/// Generates wait durations for exponential backoff according to
155+
/// [RetrySettings].
156+
///
157+
/// Uses [clock] to enforce [totalTimeout].
158+
@internal
159+
Iterable<Duration> delaySequence({
160+
int? maxRetries,
161+
Duration? totalTimeout = const Duration(minutes: 1),
162+
required Duration initialDelay,
163+
required Duration maxDelay,
164+
required double delayMultiplier,
165+
Clock clock = const Clock(),
166+
Random? random,
167+
}) sync* {
168+
final noRetriesAfter = totalTimeout == null
169+
? null
170+
: clock.fromNowBy(totalTimeout);
171+
final rng = random ?? Random();
172+
var reachedMax = false;
173+
for (var i = 0; (maxRetries == null) || (i < maxRetries); i++) {
174+
if (noRetriesAfter != null && clock.now().isAfter(noRetriesAfter)) {
175+
break;
176+
}
177+
final baseDelay = reachedMax
178+
? maxDelay
179+
: initialDelay * pow(delayMultiplier, i);
180+
if (!reachedMax && baseDelay >= maxDelay) {
181+
reachedMax = true;
182+
}
183+
final effectiveBase = reachedMax ? maxDelay : baseDelay;
184+
final jitterFactor = 0.8 + 0.4 * rng.nextDouble();
185+
yield Duration(
186+
microseconds: (effectiveBase.inMicroseconds * jitterFactor).round(),
187+
);
188+
}
189+
}
190+
191+
/// Returns whether [e] is considered a retryable error.
192+
@internal
193+
bool isRetryable(Object e) {
194+
if (e is! Exception) return false;
195+
return switch (e) {
196+
GrpcError(:final code) => switch (code) {
197+
StatusCode.aborted ||
198+
StatusCode.deadlineExceeded ||
199+
StatusCode.internal ||
200+
StatusCode.resourceExhausted ||
201+
StatusCode.unavailable ||
202+
StatusCode.unknown => true,
203+
_ => false,
204+
},
205+
ConflictException(:final status) when status?.code == StatusCode.aborted =>
206+
true,
207+
ServiceException(:final status) when status?.code == StatusCode.aborted =>
208+
true,
209+
BadGatewayException() ||
210+
RequestTimeoutException() ||
211+
ServiceUnavailableException() ||
212+
GatewayTimeoutException() ||
213+
TooManyRequestsException() => true,
214+
InternalServerErrorException(:final status) =>
215+
status?.code != StatusCode.dataLoss,
216+
_ => false,
217+
};
218+
}
219+
220+
/// Runs [body] with exponential backoff retries.
221+
///
222+
/// Only transient gRPC errors and retryable [ServiceException]s are retried.
223+
/// If [isIdempotent] is `false`, the operation is never retried and the first
224+
/// error is rethrown.
225+
Future<T> runWithRetry<T>(
226+
Future<T> Function() body, {
227+
required RetrySettings settings,
228+
required bool isIdempotent,
229+
}) async {
230+
final delays = delaySequence(
231+
maxRetries: settings.maxRetries,
232+
totalTimeout: settings.totalTimeout,
233+
initialDelay: settings.initialDelay,
234+
maxDelay: settings.maxDelay,
235+
delayMultiplier: settings.delayMultiplier,
236+
).iterator;
237+
238+
while (true) {
239+
try {
240+
return await body();
241+
} on Exception catch (e) {
242+
if (!isIdempotent || !isRetryable(e)) rethrow;
243+
244+
if (delays.moveNext()) {
245+
await Future<void>.delayed(delays.current);
246+
} else {
247+
rethrow;
248+
}
249+
}
250+
}
251+
}

‎pkgs/google_cloud_pubsub/pubspec.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ environment:
1111
resolution: workspace
1212

1313
dependencies:
14+
clock: ^1.1.0
1415
ffi: ^2.2.0
1516
fixnum: ^1.1.1
1617
google_cloud_rpc: ^0.6.0

0 commit comments

Comments
 (0)