Skip to content

Retry transient GitHub REST read failures #82

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dbachko merged 2 commits into from
Apr 2, 2026
Merged

Retry transient GitHub REST read failures #82

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bdb550b
fix(github): retry transient REST read failures
dbachko Apr 1, 2026
c3af76e
fix(github): cover retry EOF edge cases
dbachko Apr 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
282 changes: 256 additions & 26 deletions rust/src/github.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,19 @@ 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};
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<Vec<Issue>>;
Expand Down Expand Up @@ -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::<JsonValue>()
.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::<JsonValue>().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<T: DeserializeOwned>(&self, query: &str, variables: JsonValue) -> Result<T> {
Expand Down Expand Up @@ -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::<reqwest::Error>()
.map(|error| {
error.is_timeout() || error.is_connect() || error.is_request() || error.is_body()
})
.unwrap_or(false)
|| cause
.downcast_ref::<serde_json::Error>()
.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))
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

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;
Expand Down Expand Up @@ -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};

Expand All @@ -1836,6 +1928,28 @@ mod tests {
Settings::from_workflow(&definition).unwrap()
}

fn accept_with_deadline(
listener: &TcpListener,
deadline: Instant,
) -> std::io::Result<TcpStream> {
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;
Expand Down Expand Up @@ -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();
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

#[tokio::test]
async fn projects_v2_candidate_fetch_normalizes_issue_fields() {
let server = MockServer::start().await;
Expand Down
Loading