//! Retrying a request whose server went away, with a bounded, jittered backoff. //! //! Retrying is safe because nothing is recorded until a completion is final: a retry sends the same //! bytes again. Only errors that mean "the server went away" are retried, and only within the first //! window and attempt budget, so a dead server is not chased forever. use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use super::{ChatEvent, ChatRequest, Client, Completion, InferError}; pub fn is_retryable(e: &InferError) -> bool { match e { InferError::Connect(_) => true, // the server went away and may be back InferError::StreamClosedEarly => true, InferError::Stalled => true, InferError::LoadTimeout => true, InferError::Http { status: 503, .. } => true, InferError::Busy => false, // the slot is busy; the gate already said so InferError::WaitTimeout => false, // it has already waited InferError::ThinkingOverrun => false, InferError::Protocol(_) => false, InferError::Http { .. } => false, // any other status would fail again } } pub fn backoff_ms(schedule: &[u64], retry: u32, jitter: i32) -> u64 { // Retry numbers start at 1; 0 is treated as 1. let retry = retry.max(1); // The base is schedule[retry - 1], or the last entry past the end, or 0 for an empty schedule. let index = usize::try_from(retry - 1).unwrap_or(usize::MAX); let base = schedule .get(index) .copied() .or_else(|| schedule.last().copied()) .unwrap_or(0); // Jitter is clamped to -1000..=1000 and moves the wait by up to a quarter of the base. let jitter = jitter.clamp(-1000, 1000); let value = i128::from(base) + i128::from(base / 4) * i128::from(jitter) / 1000; u64::try_from(value.clamp(0, i128::from(u64::MAX))).unwrap_or(u64::MAX) } impl Client { pub fn chat_with_retry( &self, req: &ChatRequest, on_event: &mut dyn FnMut(&ChatEvent), ) -> Result { let start = Instant::now(); let mut attempt: u32 = 1; loop { match self.chat(req, on_event) { Ok(completion) => return Ok(completion), Err(e) => { if !is_retryable(&e) || attempt >= self.cfg.limits.retry_attempts { return Err(e); } let wait = backoff_ms(&self.cfg.limits.retry_backoff_ms, attempt, now_jitter()); if elapsed_ms(start) + wait > self.cfg.limits.retry_window_ms { return Err(e); } on_event(&ChatEvent::Retrying { attempt: attempt + 1, after_ms: wait, error: e.to_string(), }); std::thread::sleep(Duration::from_millis(wait)); attempt += 1; } } } } } fn now_jitter() -> i32 { let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|since| since.subsec_nanos()) .unwrap_or(0); // A nanosecond count reduced to -1000..=1000 is random enough for jitter. i32::try_from(nanos % 2000).unwrap_or(999) - 1000 } fn elapsed_ms(start: Instant) -> u64 { start.elapsed().as_millis().try_into().unwrap_or(u64::MAX) }