From bdb550bd0ad6a3e7412905e74af25971981d7692 Mon Sep 17 00:00:00 2001 From: Dzmitry Bachko Date: Wed, 1 Apr 2026 16:47:27 -0700 Subject: [PATCH 1/2] fix(github): retry transient REST read failures Summary: - retry GitHub REST GET reads when request send or JSON body decode fails with transient transport errors - keep non-GET requests single-shot and preserve existing rate-limit handling - add a regression test that simulates a truncated JSON response and verifies the second GET succeeds Rationale: - transient connection resets and truncated bodies can fail otherwise healthy GitHub reads - limiting retries to idempotent GET requests reduces recovery noise without risking duplicate writes Tests: - cargo test rest_json_retries_truncated_get_responses --- rust/src/github.rs | 192 +++++++++++++++++++++++++++++++++++++++------ 1 file changed, 166 insertions(+), 26 deletions(-) diff --git a/rust/src/github.rs b/rust/src/github.rs index 06c79c8..1cb6beb 100644 --- a/rust/src/github.rs +++ b/rust/src/github.rs @@ -9,6 +9,7 @@ use reqwest::Client; use serde::de::DeserializeOwned; use serde::Deserialize; 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 +17,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 +157,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 +1492,47 @@ 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) + || 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", + "broken pipe", + "request or response body error", + "error sending request for url", + "client error (sendrequest)", + "channel closed", + "end of file before message length reached", + ] + .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 +1890,10 @@ fn issue_locator(issue: &Issue) -> Result<(String, String, u64)> { #[cfg(test)] mod tests { + use std::io::{Read, Write}; + use std::net::{Shutdown, TcpListener}; + use std::thread; + use wiremock::matchers::{body_string_contains, method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -1876,6 +1960,62 @@ 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 (mut socket, _) = listener.accept().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, _) = listener.accept().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; From c3af76e9f003556a419e705f15b17e3f0a3c9795 Mon Sep 17 00:00:00 2001 From: Dzmitry Bachko Date: Wed, 1 Apr 2026 18:14:58 -0700 Subject: [PATCH 2/2] fix(github): cover retry EOF edge cases Summary: - retry GitHub REST GET reads when truncated JSON decode errors surface as serde_json EOF failures - harden the raw-socket retry regression test with a bounded accept deadline so CI fails cleanly instead of hanging - add coverage for both transport truncation and JSON EOF decode retry paths Rationale: - review feedback identified a decode path that could still fail before retrying a read - deadline-bounded test sockets keep retry regressions observable without risking stuck test jobs Tests: - cargo test rest_json_retries_truncated_get_responses - cargo test rest_json_retries_truncated_json_decode_errors - cargo fmt --check - cargo clippy --all-targets --all-features -- -D warnings --- rust/src/github.rs | 96 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 93 insertions(+), 3 deletions(-) diff --git a/rust/src/github.rs b/rust/src/github.rs index 1cb6beb..64958da 100644 --- a/rust/src/github.rs +++ b/rust/src/github.rs @@ -8,6 +8,7 @@ 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}; @@ -1508,6 +1509,10 @@ fn should_retry_github_read_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()) }) } @@ -1518,12 +1523,14 @@ fn retryable_github_read_error_text(message: &str) -> bool { "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)) @@ -1891,8 +1898,9 @@ fn issue_locator(issue: &Issue) -> Result<(String, String, u64)> { #[cfg(test)] mod tests { use std::io::{Read, Write}; - use std::net::{Shutdown, TcpListener}; + 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}; @@ -1920,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; @@ -1965,7 +1995,8 @@ mod tests { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let address = listener.local_addr().unwrap(); let server = thread::spawn(move || { - let (mut socket, _) = listener.accept().unwrap(); + 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 @@ -1975,7 +2006,66 @@ mod tests { .unwrap(); socket.shutdown(Shutdown::Both).unwrap(); - let (mut socket, _) = listener.accept().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}]"#;