Skip to content

Commit 716f625

Browse files
committed
feat(grpc): broaden retry policy to RESOURCE_EXHAUSTED and ABORTED
Rename `retry_on_unavailable` → `retry_on_transient` in `rust/src/retry.rs` and add a configurable `retryable_codes: HashSet<i32>` field to `RetryConfig`. Default retry set is now {UNAVAILABLE, RESOURCE_EXHAUSTED, ABORTED}: - RESOURCE_EXHAUSTED maps to HTTP 429 (rate limit) — should retry with backoff - ABORTED signals an optimistic-concurrency conflict — tonic docs recommend retry - DEADLINE_EXCEEDED is intentionally excluded — retrying without resetting the deadline just fails again immediately All 12 transport.rs callsites updated to `retry_on_transient`. New/updated tests cover the expanded code set and a custom-codes override path. Co-authored-by: P-0228
1 parent 6c4c1a9 commit 716f625

3 files changed

Lines changed: 236 additions & 151 deletions

File tree

‎AGENTS.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -283,7 +283,9 @@ except PineconeError as e:
283283
print(e) # Validation, timeout, or connection error
284284
```
285285

286-
**Retry behavior:** All HTTP methods (GET, HEAD, POST, PUT, PATCH, DELETE) are automatically retried on transient failures: connection errors (`httpx.TransportError`), 408 Request Timeout, 429 Too Many Requests (honoring `Retry-After`), and 5xx (500, 502, 503, 504). Pinecone's data-plane writes are idempotent at the server (upsert overwrites by ID, delete-by-ID is idempotent, update-by-ID is idempotent), so retrying upsert/query/fetch/delete/update on transient errors is safe. Backoff is floored full jitter: `uniform(0.1 * base, base)` where `base = min(backoff_factor**attempt, max_wait)`. Configure via `RetryConfig`.
286+
**Retry behavior (HTTP):** All HTTP methods (GET, HEAD, POST, PUT, PATCH, DELETE) are automatically retried on transient failures: connection errors (`httpx.TransportError`), 408 Request Timeout, 429 Too Many Requests (honoring `Retry-After`), and 5xx (500, 502, 503, 504). Pinecone's data-plane writes are idempotent at the server (upsert overwrites by ID, delete-by-ID is idempotent, update-by-ID is idempotent), so retrying upsert/query/fetch/delete/update on transient errors is safe. Backoff is floored full jitter: `uniform(0.1 * base, base)` where `base = min(backoff_factor**attempt, max_wait)`. Configure via `RetryConfig`.
287+
288+
**Retry behavior (gRPC):** gRPC retries on UNAVAILABLE, RESOURCE_EXHAUSTED (rate limit), and ABORTED (concurrency conflict). DEADLINE_EXCEEDED is not retried — set a longer client timeout instead. All three default retryable codes are safe for Pinecone data-plane operations (upsert, query, fetch, delete-by-id, update), which are idempotent. Backoff uses full-jitter exponential: `uniform(0, min(max_backoff, initial_backoff * multiplier^attempt))`. The set of retryable codes is configurable via `RetryConfig.retryable_codes`.
287289

288290
## Response Objects
289291

‎rust/src/retry.rs‎

Lines changed: 111 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
use std::collections::HashSet;
12
use std::future::Future;
23
use std::time::Duration;
34

@@ -15,6 +16,10 @@ pub struct RetryConfig {
1516
pub max_backoff: Duration,
1617
/// Backoff multiplier applied each attempt.
1718
pub multiplier: u32,
19+
/// gRPC status codes that trigger a retry. Defaults to UNAVAILABLE, RESOURCE_EXHAUSTED,
20+
/// ABORTED — Pinecone data-plane operations (upsert, query, fetch, delete-by-id, update)
21+
/// are idempotent and safe to retry on these transient codes.
22+
pub retryable_codes: HashSet<i32>,
1823
}
1924

2025
impl Default for RetryConfig {
@@ -24,18 +29,25 @@ impl Default for RetryConfig {
2429
initial_backoff: Duration::from_millis(100),
2530
max_backoff: Duration::from_millis(1600),
2631
multiplier: 2,
32+
retryable_codes: [
33+
tonic::Code::Unavailable as i32,
34+
tonic::Code::ResourceExhausted as i32,
35+
tonic::Code::Aborted as i32,
36+
]
37+
.into_iter()
38+
.collect(),
2739
}
2840
}
2941
}
3042

31-
/// Execute an async gRPC operation with retry on `Code::Unavailable`.
43+
/// Execute an async gRPC operation with retry on transient error codes.
3244
///
3345
/// Uses full-jitter exponential backoff:
3446
/// delay = random(0, min(max_backoff, initial_backoff * multiplier^attempt))
3547
///
36-
/// Only retries on `tonic::Code::Unavailable`. All other error codes are
37-
/// returned immediately without retry.
38-
pub async fn retry_on_unavailable<F, Fut, T>(
48+
/// Retries on any code listed in `config.retryable_codes` (default: UNAVAILABLE,
49+
/// RESOURCE_EXHAUSTED, ABORTED). All other error codes are returned immediately without retry.
50+
pub async fn retry_on_transient<F, Fut, T>(
3951
config: &RetryConfig,
4052
mut operation: F,
4153
) -> Result<T, Status>
@@ -48,7 +60,7 @@ where
4860
loop {
4961
match operation().await {
5062
Ok(val) => return Ok(val),
51-
Err(status) if status.code() == tonic::Code::Unavailable => {
63+
Err(status) if config.retryable_codes.contains(&(status.code() as i32)) => {
5264
if attempt >= config.max_retries {
5365
return Err(status);
5466
}
@@ -62,6 +74,8 @@ where
6274
let ms = rand::rng().random_range(0..=capped.as_millis() as u64);
6375
Duration::from_millis(ms)
6476
};
77+
// TODO: honor server-supplied pushback hints (grpc-retry-pushback-ms trailer
78+
// or Retry-After in trailers) before falling back to computed jitter backoff.
6579
tokio::time::sleep(jittered).await;
6680
attempt += 1;
6781
}
@@ -82,6 +96,7 @@ mod tests {
8296
initial_backoff: Duration::from_millis(1), // fast for tests
8397
max_backoff: Duration::from_millis(10),
8498
multiplier: 2,
99+
..Default::default()
85100
}
86101
}
87102

@@ -91,7 +106,7 @@ mod tests {
91106
let count = call_count.clone();
92107

93108
let config = test_config(5);
94-
let result = retry_on_unavailable(&config, || {
109+
let result = retry_on_transient(&config, || {
95110
let count = count.clone();
96111
async move {
97112
let n = count.fetch_add(1, Ordering::SeqCst);
@@ -116,7 +131,7 @@ mod tests {
116131
let count = call_count.clone();
117132

118133
let config = test_config(5);
119-
let result = retry_on_unavailable(&config, || {
134+
let result = retry_on_transient(&config, || {
120135
let count = count.clone();
121136
async move {
122137
count.fetch_add(1, Ordering::SeqCst);
@@ -132,32 +147,110 @@ mod tests {
132147
}
133148

134149
#[tokio::test]
135-
async fn no_retry_on_resource_exhausted() {
150+
async fn retry_occurs_on_resource_exhausted() {
136151
let call_count = Arc::new(AtomicU32::new(0));
137152
let count = call_count.clone();
138153

139154
let config = test_config(5);
140-
let result = retry_on_unavailable(&config, || {
155+
let result = retry_on_transient(&config, || {
156+
let count = count.clone();
157+
async move {
158+
let n = count.fetch_add(1, Ordering::SeqCst);
159+
if n < 2 {
160+
Err(Status::resource_exhausted("rate limited"))
161+
} else {
162+
Ok::<(), Status>(())
163+
}
164+
}
165+
})
166+
.await;
167+
168+
assert!(result.is_ok());
169+
// Initial call + 2 retries = 3 total calls
170+
assert_eq!(call_count.load(Ordering::SeqCst), 3);
171+
}
172+
173+
#[tokio::test]
174+
async fn retry_occurs_on_aborted() {
175+
let call_count = Arc::new(AtomicU32::new(0));
176+
let count = call_count.clone();
177+
178+
let config = test_config(5);
179+
let result = retry_on_transient(&config, || {
180+
let count = count.clone();
181+
async move {
182+
let n = count.fetch_add(1, Ordering::SeqCst);
183+
if n < 2 {
184+
Err(Status::aborted("conflict"))
185+
} else {
186+
Ok::<(), Status>(())
187+
}
188+
}
189+
})
190+
.await;
191+
192+
assert!(result.is_ok());
193+
// Initial call + 2 retries = 3 total calls
194+
assert_eq!(call_count.load(Ordering::SeqCst), 3);
195+
}
196+
197+
#[tokio::test]
198+
async fn no_retry_on_internal() {
199+
let call_count = Arc::new(AtomicU32::new(0));
200+
let count = call_count.clone();
201+
202+
let config = test_config(5);
203+
let result = retry_on_transient(&config, || {
141204
let count = count.clone();
142205
async move {
143206
count.fetch_add(1, Ordering::SeqCst);
144-
Err::<(), Status>(Status::resource_exhausted("rate limited"))
207+
Err::<(), Status>(Status::internal("oops"))
145208
}
146209
})
147210
.await;
148211

149212
assert!(result.is_err());
150-
assert_eq!(result.unwrap_err().code(), tonic::Code::ResourceExhausted);
213+
assert_eq!(result.unwrap_err().code(), tonic::Code::Internal);
151214
assert_eq!(call_count.load(Ordering::SeqCst), 1);
152215
}
153216

217+
#[tokio::test]
218+
async fn custom_retryable_codes_override() {
219+
let call_count = Arc::new(AtomicU32::new(0));
220+
let count = call_count.clone();
221+
222+
let config = RetryConfig {
223+
max_retries: 3,
224+
initial_backoff: Duration::from_millis(1),
225+
max_backoff: Duration::from_millis(10),
226+
multiplier: 2,
227+
retryable_codes: HashSet::from([tonic::Code::DeadlineExceeded as i32]),
228+
};
229+
let result = retry_on_transient(&config, || {
230+
let count = count.clone();
231+
async move {
232+
let n = count.fetch_add(1, Ordering::SeqCst);
233+
if n < 2 {
234+
Err(Status::deadline_exceeded("timeout"))
235+
} else {
236+
Ok::<(), Status>(())
237+
}
238+
}
239+
})
240+
.await;
241+
242+
assert!(result.is_ok());
243+
// DEADLINE_EXCEEDED is retried under this custom config
244+
assert_eq!(call_count.load(Ordering::SeqCst), 3);
245+
}
246+
154247
#[tokio::test]
155248
async fn respects_max_retry_count() {
156249
let call_count = Arc::new(AtomicU32::new(0));
157250
let count = call_count.clone();
158251

159252
let config = test_config(3);
160-
let result = retry_on_unavailable(&config, || {
253+
let result = retry_on_transient(&config, || {
161254
let count = count.clone();
162255
async move {
163256
count.fetch_add(1, Ordering::SeqCst);
@@ -178,7 +271,7 @@ mod tests {
178271
let count = call_count.clone();
179272

180273
let config = test_config(0);
181-
let result = retry_on_unavailable(&config, || {
274+
let result = retry_on_transient(&config, || {
182275
let count = count.clone();
183276
async move {
184277
count.fetch_add(1, Ordering::SeqCst);
@@ -200,23 +293,25 @@ mod tests {
200293
initial_backoff: Duration::from_millis(10),
201294
max_backoff: Duration::from_millis(500),
202295
multiplier: 2,
296+
..Default::default()
203297
};
204298
let config_3 = RetryConfig {
205299
max_retries: 3,
206300
initial_backoff: Duration::from_millis(10),
207301
max_backoff: Duration::from_millis(500),
208302
multiplier: 2,
303+
..Default::default()
209304
};
210305

211306
let start_1 = std::time::Instant::now();
212-
let _ = retry_on_unavailable(&config_1, || async {
307+
let _ = retry_on_transient(&config_1, || async {
213308
Err::<(), Status>(Status::unavailable("unavailable"))
214309
})
215310
.await;
216311
let elapsed_1 = start_1.elapsed();
217312

218313
let start_3 = std::time::Instant::now();
219-
let _ = retry_on_unavailable(&config_3, || async {
314+
let _ = retry_on_transient(&config_3, || async {
220315
Err::<(), Status>(Status::unavailable("unavailable"))
221316
})
222317
.await;
@@ -233,7 +328,7 @@ mod tests {
233328
async fn success_on_first_attempt_returns_immediately() {
234329
let config = test_config(5);
235330
let result =
236-
retry_on_unavailable(&config, || async { Ok::<&str, Status>("immediate") }).await;
331+
retry_on_transient(&config, || async { Ok::<&str, Status>("immediate") }).await;
237332

238333
assert!(result.is_ok());
239334
assert_eq!(result.unwrap(), "immediate");

0 commit comments

Comments
 (0)