diff --git a/rust/src/github.rs b/rust/src/github.rs index 06c79c8..64958da 100644 --- a/rust/src/github.rs +++ b/rust/src/github.rs @@ -8,7 +8,9 @@ use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, AUTHORIZATION, USER_AGENT} use reqwest::Client; use serde::de::DeserializeOwned; use serde::Deserialize; +use serde_json::error::Category as JsonErrorCategory; use serde_json::{json, Value as JsonValue}; +use tokio::time::{sleep, Duration}; use tracing::{debug, warn}; use crate::config::{normalize_issue_state, FieldSourceType, GitHubMode, TrackerSettings}; @@ -16,6 +18,9 @@ use crate::github_bootstrap::derive_status_option_names; use crate::model::{BlockerRef, Issue}; use crate::providers::is_workpad_comment; +const GITHUB_READ_MAX_ATTEMPTS: u32 = 3; +const GITHUB_READ_RETRY_BASE_DELAY_MS: u64 = 250; + #[async_trait] pub trait Tracker: Send + Sync { async fn fetch_candidate_issues(&self) -> Result>; @@ -153,36 +158,71 @@ impl GitHubTracker { self.settings.rest_endpoint.trim_end_matches('/'), path ); - let request = self.client.request(method, url); - let request = if let Some(body) = body { - request.json(&body) - } else { - request - }; - let response = request - .send() - .await - .with_context(|| format!("failed to send GitHub REST request for {path}"))?; - - let status = response.status(); - let body = response - .json::() - .await - .context("failed to decode GitHub REST response")?; + let mut attempt = 1_u32; + loop { + let request = self.client.request(method.clone(), &url); + let request = if let Some(body) = body.as_ref() { + request.json(body) + } else { + request + }; + + let response = match request.send().await { + Ok(response) => response, + Err(error) => { + let error = anyhow!(error) + .context(format!("failed to send GitHub REST request for {path}")); + if should_retry_github_read_error(&method, &error, attempt) { + warn!( + path, + attempt, + max_attempts = GITHUB_READ_MAX_ATTEMPTS, + error = ?error, + "GitHub REST read failed; retrying" + ); + sleep(github_read_retry_delay(attempt)).await; + attempt += 1; + continue; + } + return Err(error); + } + }; + + let status = response.status(); + let body = match response.json::().await { + Ok(body) => body, + Err(error) => { + let error = anyhow!(error).context("failed to decode GitHub REST response"); + if should_retry_github_read_error(&method, &error, attempt) { + warn!( + path, + attempt, + max_attempts = GITHUB_READ_MAX_ATTEMPTS, + error = ?error, + "GitHub REST read body failed; retrying" + ); + sleep(github_read_retry_delay(attempt)).await; + attempt += 1; + continue; + } + return Err(error); + } + }; - if !status.is_success() { - if rest_response_indicates_rate_limit(status.as_u16(), &body) { - return Err(anyhow!("github_rate_limit")); + if !status.is_success() { + if rest_response_indicates_rate_limit(status.as_u16(), &body) { + return Err(anyhow!("github_rate_limit")); + } + return Err(anyhow!( + "github_rest_status: {} path={} body={body}", + status, + path + )); } - return Err(anyhow!( - "github_rest_status: {} path={} body={body}", - status, - path - )); - } - Ok(body) + return Ok(body); + } } async fn graphql(&self, query: &str, variables: JsonValue) -> Result { @@ -1453,6 +1493,53 @@ pub fn is_rate_limited_error(error: &anyhow::Error) -> bool { .any(|cause| cause.to_string().contains("github_rate_limit")) } +fn should_retry_github_read_error( + method: &reqwest::Method, + error: &anyhow::Error, + attempt: u32, +) -> bool { + if method != reqwest::Method::GET || attempt >= GITHUB_READ_MAX_ATTEMPTS { + return false; + } + + error.chain().any(|cause| { + cause + .downcast_ref::() + .map(|error| { + error.is_timeout() || error.is_connect() || error.is_request() || error.is_body() + }) + .unwrap_or(false) + || cause + .downcast_ref::() + .map(|error| matches!(error.classify(), JsonErrorCategory::Eof)) + .unwrap_or(false) + || retryable_github_read_error_text(&cause.to_string()) + }) +} + +fn retryable_github_read_error_text(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + [ + "connection closed before message completed", + "connection reset", + "unexpected eof", + "unexpected end of file", + "broken pipe", + "request or response body error", + "error sending request for url", + "client error (sendrequest)", + "channel closed", + "end of file before message length reached", + "eof while parsing", + ] + .into_iter() + .any(|needle| message.contains(needle)) +} + +fn github_read_retry_delay(attempt: u32) -> Duration { + Duration::from_millis(GITHUB_READ_RETRY_BASE_DELAY_MS.saturating_mul(1 << (attempt - 1))) +} + fn rest_response_indicates_rate_limit(status: u16, body: &JsonValue) -> bool { if status != 403 && status != 429 { return false; @@ -1810,6 +1897,11 @@ fn issue_locator(issue: &Issue) -> Result<(String, String, u64)> { #[cfg(test)] mod tests { + use std::io::{Read, Write}; + use std::net::{Shutdown, TcpListener, TcpStream}; + use std::thread; + use std::time::{Duration as StdDuration, Instant}; + use wiremock::matchers::{body_string_contains, method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -1836,6 +1928,28 @@ mod tests { Settings::from_workflow(&definition).unwrap() } + fn accept_with_deadline( + listener: &TcpListener, + deadline: Instant, + ) -> std::io::Result { + listener.set_nonblocking(true)?; + loop { + match listener.accept() { + Ok((socket, _)) => return Ok(socket), + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + if Instant::now() >= deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "timed out waiting for test connection", + )); + } + thread::sleep(StdDuration::from_millis(10)); + } + Err(error) => return Err(error), + } + } + } + #[tokio::test] async fn project_fetch_surfaces_graphql_rate_limit_as_transient_error() { let server = MockServer::start().await; @@ -1876,6 +1990,122 @@ mod tests { assert!(is_rate_limited_error(&error)); } + #[tokio::test] + async fn rest_json_retries_truncated_get_responses() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let deadline = Instant::now() + StdDuration::from_secs(5); + let mut socket = accept_with_deadline(&listener, deadline).unwrap(); + let mut request = [0_u8; 2048]; + let _ = socket.read(&mut request); + socket + .write_all( + b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 64\r\nconnection: close\r\n\r\n[{\"id\":1", + ) + .unwrap(); + socket.shutdown(Shutdown::Both).unwrap(); + + let mut socket = accept_with_deadline(&listener, deadline).unwrap(); + let mut request = [0_u8; 2048]; + let _ = socket.read(&mut request); + let body = r#"[{"ok":true}]"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + socket.write_all(response.as_bytes()).unwrap(); + socket.shutdown(Shutdown::Both).unwrap(); + }); + + let tracker = GitHubTracker::new( + settings(&format!( + r#"tracker: + kind: github + owner: openai + repo: kairastra + api_key: fake + endpoint: http://{0}/graphql + rest_endpoint: http://{0} + mode: issues_only + status_source: + type: label +"#, + address + )) + .tracker, + ) + .unwrap(); + + let response = tracker + .rest_json(reqwest::Method::GET, "/repos/openai/kairastra/issues", None) + .await + .unwrap(); + assert_eq!(response, serde_json::json!([{ "ok": true }])); + + server.join().unwrap(); + } + + #[tokio::test] + async fn rest_json_retries_truncated_json_decode_errors() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let deadline = Instant::now() + StdDuration::from_secs(5); + let mut socket = accept_with_deadline(&listener, deadline).unwrap(); + let mut request = [0_u8; 2048]; + let _ = socket.read(&mut request); + let body = r#"[{"ok":true"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + socket.write_all(response.as_bytes()).unwrap(); + socket.shutdown(Shutdown::Both).unwrap(); + + let mut socket = accept_with_deadline(&listener, deadline).unwrap(); + let mut request = [0_u8; 2048]; + let _ = socket.read(&mut request); + let body = r#"[{"ok":true}]"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + socket.write_all(response.as_bytes()).unwrap(); + socket.shutdown(Shutdown::Both).unwrap(); + }); + + let tracker = GitHubTracker::new( + settings(&format!( + r#"tracker: + kind: github + owner: openai + repo: kairastra + api_key: fake + endpoint: http://{0}/graphql + rest_endpoint: http://{0} + mode: issues_only + status_source: + type: label +"#, + address + )) + .tracker, + ) + .unwrap(); + + let response = tracker + .rest_json(reqwest::Method::GET, "/repos/openai/kairastra/issues", None) + .await + .unwrap(); + assert_eq!(response, serde_json::json!([{ "ok": true }])); + + server.join().unwrap(); + } + #[tokio::test] async fn projects_v2_candidate_fetch_normalizes_issue_fields() { let server = MockServer::start().await;