Skip to content

Commit e2300cc

Browse files
Bound registry downloads by ApiTimeouts instead of a 60 s total deadline (#872) (#876)
* Start refactor for #872 Assisted-by: Claude Code:claude-opus-5-5 * Bound registry fetches by ApiTimeouts Hosted upstream restore (rollback, vendor takeover) and vendored Maven built their registry clients with a 60 s whole-request deadline and no connect bound, so on a slow link an original tarball, module zip or jar that took over a minute to download failed with a transport error even while bytes were still arriving, and a black-holed host held the run for the full minute. Both now build through one registry_fetch::registry_client_builder that applies the shared ApiTimeouts policy: 10 s connect plus 60 s of silence, no total deadline. registry_fetch::download reads the body through utils::http::read_capped, deleting the last hand-rolled copy of the capped reader; the cap and its checks are unchanged, only the refusal wording now matches the other capped downloads. Assisted-by: Claude Code:claude-opus-5-5 * Route the JVM jar digests through utils::digest #646 landed inline sha1/sha256 computations in patch/jvm_jar.rs, patch/sidecars/maven.rs and crawlers/gradle_cache.rs after the utils::digest ratchet, so production_digests_go_through_the_helpers fails on main. jvm_jar's private sha1_hex/sha256_hex copies and the Maven sidecar's inline sha1 now call the shared helpers; gradle_cache.rs, which an open PR also edits, joins the pending list for now. Hashes are byte-identical. Assisted-by: Claude Code:claude-opus-5-5 * Drop stale entries from PENDING_INLINE_DIGESTS main's gradle_cache.rs, jvm_jar.rs and sidecars/maven.rs already route their digests through utils::digest, but the ratchet still lists them as pending, so production_digests_go_through_the_helpers fails on a stale entry. Remove the three entries so the ratchet matches the code. Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 1c3dcac commit e2300cc

2 files changed

Lines changed: 154 additions & 33 deletions

File tree

‎crates/socket-patch-core/src/vendor/maven_repo.rs‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
use std::collections::HashMap;
22
use std::path::{Path, PathBuf};
3-
use std::time::Duration;
43

54
use serde_json::Value;
65
use sha1::Sha1;
@@ -1822,9 +1821,7 @@ async fn fetch_pom_bytes(url: &str) -> Result<Vec<u8>, String> {
18221821
}
18231822

18241823
pub(crate) async fn fetch_registry_bytes(url: &str, cap: u64) -> Result<Vec<u8>, String> {
1825-
let client = reqwest::Client::builder()
1826-
.user_agent(MAVEN_USER_AGENT)
1827-
.timeout(Duration::from_secs(60))
1824+
let client = super::registry_fetch::registry_client_builder(MAVEN_USER_AGENT)
18281825
.build()
18291826
.map_err(|e| format!("build http client: {e}"))?;
18301827
let resp = client

‎crates/socket-patch-core/src/vendor/registry_fetch.rs‎

Lines changed: 153 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
//! Bounded archive readers, integrity verification and registry metadata transport.
22
33
use std::path::{Path, PathBuf};
4-
use std::time::Duration;
54

65
use base64::Engine as _;
76
use sha1::Sha1;
87
use sha2::{Digest, Sha256, Sha384, Sha512};
98

9+
use crate::api::retry::ApiTimeouts;
1010
use crate::constants::USER_AGENT;
1111
use crate::patch::apply::is_safe_relative_subpath;
1212

@@ -43,13 +43,61 @@ pub enum FetchError {
4343
pub type RegistryClient = reqwest::Client;
4444

4545
pub fn build_registry_client() -> RegistryClient {
46-
reqwest::Client::builder()
47-
.user_agent(USER_AGENT)
48-
.timeout(Duration::from_secs(60))
46+
registry_client_builder(USER_AGENT)
4947
.build()
5048
.unwrap_or_else(|_| reqwest::Client::new())
5149
}
5250

51+
/// The one builder behind every registry client (npm-family, PyPI, Go,
52+
/// NuGet, Maven), sending `user_agent`. It applies the shared
53+
/// [`ApiTimeouts`] transport policy: a connect bound plus an idle read
54+
/// bound that restarts on every chunk, and no total deadline, so a large
55+
/// artifact that keeps streaming is never cut off while a stalled host
56+
/// still fails the fetch.
57+
pub(crate) fn registry_client_builder(user_agent: &str) -> reqwest::ClientBuilder {
58+
registry_timeouts().apply(reqwest::Client::builder().user_agent(user_agent))
59+
}
60+
61+
fn registry_timeouts() -> ApiTimeouts {
62+
#[cfg(test)]
63+
if let Some(t) = test_timeouts::get() {
64+
return t;
65+
}
66+
ApiTimeouts::default()
67+
}
68+
69+
/// Test-only override of [`registry_timeouts`] for the current thread, so a
70+
/// test can prove the idle bound and the absence of a total deadline in
71+
/// seconds rather than minutes. `#[tokio::test]` runs on one thread.
72+
#[cfg(test)]
73+
pub(crate) mod test_timeouts {
74+
use std::cell::Cell;
75+
76+
use crate::api::retry::ApiTimeouts;
77+
78+
thread_local! {
79+
static OVERRIDE: Cell<Option<ApiTimeouts>> = const { Cell::new(None) };
80+
}
81+
82+
pub(crate) fn get() -> Option<ApiTimeouts> {
83+
OVERRIDE.with(Cell::get)
84+
}
85+
86+
/// Shortens the bounds until the returned guard drops.
87+
pub(crate) fn set(t: ApiTimeouts) -> Guard {
88+
OVERRIDE.with(|c| c.set(Some(t)));
89+
Guard
90+
}
91+
92+
pub(crate) struct Guard;
93+
94+
impl Drop for Guard {
95+
fn drop(&mut self) {
96+
OVERRIDE.with(|c| c.set(None));
97+
}
98+
}
99+
}
100+
53101
/// The npm registry base after the env override.
54102
pub fn npm_registry_base() -> String {
55103
std::env::var("SOCKET_NPM_REGISTRY")
@@ -1170,14 +1218,14 @@ fn walk_zip_with_prefix(
11701218
Ok(())
11711219
}
11721220

1173-
/// Capped download. http(s) only; the cap is enforced on the declared
1174-
/// Content-Length AND the actual stream (a lying server cannot blow past
1175-
/// it).
1221+
/// Capped download. http(s) only; [`crate::utils::http::read_capped`]
1222+
/// enforces [`MAX_DOWNLOAD_BYTES`] on the declared Content-Length AND the
1223+
/// actual stream (a lying server cannot blow past it).
11761224
pub(crate) async fn download(client: &reqwest::Client, url: &str) -> Result<Vec<u8>, String> {
11771225
if !(url.starts_with("https://") || url.starts_with("http://")) {
11781226
return Err(format!("refusing non-http(s) artifact URL `{url}`"));
11791227
}
1180-
let mut resp = client
1228+
let resp = client
11811229
.get(url)
11821230
.send()
11831231
.await
@@ -1186,27 +1234,9 @@ pub(crate) async fn download(client: &reqwest::Client, url: &str) -> Result<Vec<
11861234
if !status.is_success() {
11871235
return Err(format!("GET {url}: HTTP {status}"));
11881236
}
1189-
if let Some(len) = resp.content_length() {
1190-
if len > MAX_DOWNLOAD_BYTES {
1191-
return Err(format!(
1192-
"{url}: artifact is {len} bytes (cap {MAX_DOWNLOAD_BYTES})"
1193-
));
1194-
}
1195-
}
1196-
let mut bytes: Vec<u8> = Vec::new();
1197-
while let Some(chunk) = resp
1198-
.chunk()
1237+
crate::utils::http::read_capped(resp, MAX_DOWNLOAD_BYTES, "registry artifact")
11991238
.await
1200-
.map_err(|e| format!("reading {url}: {e}"))?
1201-
{
1202-
if bytes.len() as u64 + chunk.len() as u64 > MAX_DOWNLOAD_BYTES {
1203-
return Err(format!(
1204-
"{url}: artifact exceeds the {MAX_DOWNLOAD_BYTES}-byte cap"
1205-
));
1206-
}
1207-
bytes.extend_from_slice(&chunk);
1208-
}
1209-
Ok(bytes)
1239+
.map_err(|e| format!("{url}: {e}"))
12101240
}
12111241

12121242
/// Verify archive bytes against lock-recorded integrity. Berry cache checksums
@@ -2993,12 +3023,106 @@ mod tests {
29933023
.await
29943024
.unwrap_err();
29953025
assert!(
2996-
err.contains("exceeds the") && err.contains("cap"),
3026+
err.contains("exceeded") && err.contains("cap"),
29973027
"the stream cap must fire without a Content-Length: {err}"
29983028
);
29993029
server.abort();
30003030
}
30013031

3032+
/// Serves one GET per accepted connection: a 200 head declaring
3033+
/// `chunks × 1 KiB`, then each 1 KiB chunk after its `gaps` delay.
3034+
async fn paced_server(
3035+
gaps: Vec<std::time::Duration>,
3036+
) -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
3037+
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
3038+
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
3039+
let addr = listener.local_addr().unwrap();
3040+
let server = tokio::spawn(async move {
3041+
loop {
3042+
let Ok((mut sock, _)) = listener.accept().await else {
3043+
return;
3044+
};
3045+
let gaps = gaps.clone();
3046+
tokio::spawn(async move {
3047+
let mut buf = [0u8; 4096];
3048+
let _ = sock.read(&mut buf).await; // request head
3049+
let head = format!(
3050+
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
3051+
gaps.len() * 1024
3052+
);
3053+
if sock.write_all(head.as_bytes()).await.is_err() {
3054+
return;
3055+
}
3056+
for gap in gaps {
3057+
tokio::time::sleep(gap).await;
3058+
if sock.write_all(&[b'x'; 1024]).await.is_err() {
3059+
return;
3060+
}
3061+
}
3062+
});
3063+
}
3064+
});
3065+
(addr, server)
3066+
}
3067+
3068+
/// Every registry client: hosted upstream restore's
3069+
/// `build_registry_client` + `download`, and vendored Maven's
3070+
/// `fetch_registry_bytes` (which sends Maven's own user agent).
3071+
async fn fetch_through_every_registry_client(url: &str) -> Vec<Result<Vec<u8>, String>> {
3072+
vec![
3073+
download(&build_registry_client(), url).await,
3074+
crate::vendor::maven_repo::fetch_registry_bytes(url, MAX_DOWNLOAD_BYTES).await,
3075+
]
3076+
}
3077+
3078+
const SHORT_BOUNDS: ApiTimeouts = ApiTimeouts {
3079+
connect: std::time::Duration::from_secs(2),
3080+
read: std::time::Duration::from_millis(400),
3081+
};
3082+
3083+
#[tokio::test]
3084+
async fn registry_clients_have_no_total_deadline() {
3085+
// #872: the registry clients set a 60 s whole-request deadline, so
3086+
// a slow but steady download was aborted mid-body. Under the shared
3087+
// `ApiTimeouts` policy only silence counts: a body that trickles
3088+
// for 4× the (shortened) idle bound, never pausing that long,
3089+
// arrives whole through every registry client.
3090+
let _bounds = test_timeouts::set(SHORT_BOUNDS);
3091+
let gaps = vec![std::time::Duration::from_millis(100); 16];
3092+
let (addr, server) = paced_server(gaps).await;
3093+
let url = format!("http://{addr}/slow.tgz");
3094+
for got in fetch_through_every_registry_client(&url).await {
3095+
assert_eq!(
3096+
got.expect("a progressing body must not time out").len(),
3097+
16 * 1024
3098+
);
3099+
}
3100+
server.abort();
3101+
}
3102+
3103+
#[tokio::test]
3104+
async fn registry_clients_fail_a_body_that_stalls_past_the_idle_bound() {
3105+
// A connection that goes silent mid-body fails at the idle bound
3106+
// instead of holding the run until the server resumes (here 5 s
3107+
// later; on `main` both clients waited it out and succeeded).
3108+
let _bounds = test_timeouts::set(SHORT_BOUNDS);
3109+
let mut gaps = vec![std::time::Duration::ZERO; 4];
3110+
gaps.push(std::time::Duration::from_secs(5));
3111+
let (addr, server) = paced_server(gaps).await;
3112+
let url = format!("http://{addr}/stall.tgz");
3113+
let started = std::time::Instant::now();
3114+
for got in fetch_through_every_registry_client(&url).await {
3115+
let err = got.expect_err("a stalled body must fail at the idle bound");
3116+
assert!(err.contains("error reading"), "{err}");
3117+
}
3118+
assert!(
3119+
started.elapsed() < std::time::Duration::from_secs(4),
3120+
"both fetches must give up at the idle bound, took {:?}",
3121+
started.elapsed()
3122+
);
3123+
server.abort();
3124+
}
3125+
30023126
#[test]
30033127
fn total_decompressed_cap_fails_closed_across_zip_extractors() {
30043128
// The per-entry actual-bytes guards mean only HONEST content reaches

0 commit comments

Comments
 (0)