Skip to content

Commit 1c836cd

Browse files
authored
fix(fetchers): surface malformed body errors (#104)
## What Treat malformed zero-byte response bodies as fetch errors instead of successful empty responses. Closes #98. ## Why `read_body_with_timeout()` collapsed a body-stream failure before the first chunk into `(empty, false)`. That let malformed chunked responses look like successful `200 OK` fetches with empty content, and the same path could also create misleading zero-byte files under `save_to_file`. ## How - change `read_body_with_timeout()` to return `FetchError` when the body stream fails before any bytes are read - preserve partial-body behavior for mid-stream failures by returning the partial bytes as `truncated=true` - propagate the new result through the default fetcher and direct `llms.txt` docs-site path - add regressions for malformed chunked responses in both normal fetch and `save_to_file` flows ## Risk - Low - Changes only the malformed-body error path; successful responses, timeouts, size caps, and partial-body truncation stay on the existing behavior ### Checklist - [x] Unit tests are passed - [x] Smoke tests are passed - [ ] Documentation is updated - [x] Specs are up to date and not in conflict - [x] `cargo fmt --all` is passed - [x] `cargo clippy --workspace --all-targets -- -D warnings` is passed - [x] `RUSTDOCFLAGS="-D warnings" cargo doc --workspace --no-deps` is passed - [x] `cargo build --workspace --exclude fetchkit-python --release` is passed
1 parent afbc233 commit 1c836cd

3 files changed

Lines changed: 57 additions & 10 deletions

File tree

crates/fetchkit/src/fetchers/default.rs

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -292,7 +292,8 @@ impl Fetcher for DefaultFetcher {
292292

293293
// THREAT[TM-DOS-001]: Read body with timeout and size limit
294294
// THREAT[TM-DOS-003]: Size limit also protects against compressed content bombs
295-
let (body, truncated) = read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await;
295+
let (body, truncated) =
296+
read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await?;
296297
let size = body.len() as u64;
297298

298299
// Convert to string
@@ -438,7 +439,8 @@ impl Fetcher for DefaultFetcher {
438439
}
439440

440441
// Read raw body (no binary rejection for file saves)
441-
let (body, truncated) = read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await;
442+
let (body, truncated) =
443+
read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await?;
442444
let size = body.len() as u64;
443445

444446
// Save through the FileSaver
@@ -644,7 +646,7 @@ pub(crate) async fn read_body_with_timeout(
644646
response: reqwest::Response,
645647
timeout: Duration,
646648
max_size: usize,
647-
) -> (Bytes, bool) {
649+
) -> Result<(Bytes, bool), FetchError> {
648650
let mut body = Vec::new();
649651
let mut stream = response.bytes_stream();
650652
let deadline = tokio::time::Instant::now() + timeout;
@@ -660,29 +662,31 @@ pub(crate) async fn read_body_with_timeout(
660662
let remaining = max_size.saturating_sub(body.len());
661663
if remaining == 0 {
662664
warn!("Body size limit reached ({}), truncating", max_size);
663-
return (Bytes::from(body), true);
665+
return Ok((Bytes::from(body), true));
664666
}
665667
if bytes.len() > remaining {
666668
body.extend_from_slice(&bytes[..remaining]);
667669
warn!("Body size limit reached ({}), truncating", max_size);
668-
return (Bytes::from(body), true);
670+
return Ok((Bytes::from(body), true));
669671
}
670672
body.extend_from_slice(&bytes);
671673
}
672674
Some(Err(e)) => {
673675
error!("Error reading body chunk: {}", e);
674-
let has_content = !body.is_empty();
675-
return (Bytes::from(body), has_content);
676+
if body.is_empty() {
677+
return Err(FetchError::from_reqwest(e));
678+
}
679+
return Ok((Bytes::from(body), true));
676680
}
677681
None => {
678682
// Stream complete
679-
return (Bytes::from(body), false);
683+
return Ok((Bytes::from(body), false));
680684
}
681685
}
682686
}
683687
_ = timeout_future => {
684688
warn!("Body timeout reached, returning partial content");
685-
return (Bytes::from(body), true);
689+
return Ok((Bytes::from(body), true));
686690
}
687691
}
688692
}

crates/fetchkit/src/fetchers/docs_site.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -239,7 +239,7 @@ async fn fetch_llms_txt_direct(
239239
}
240240

241241
let max_body_size = options.max_body_size.unwrap_or(DEFAULT_MAX_BODY_SIZE);
242-
let (body, truncated) = read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await;
242+
let (body, truncated) = read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await?;
243243
let size = body.len() as u64;
244244
let mut content = String::from_utf8_lossy(&body).to_string();
245245

crates/fetchkit/tests/integration.rs

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@ use fetchkit::{
55
HttpMethod, LocalFileSaver, Tool,
66
};
77
use serde_json::json;
8+
use tokio::io::{AsyncReadExt, AsyncWriteExt};
9+
use tokio::net::TcpListener;
810
use tower::Service;
911
use wiremock::matchers::{method, path};
1012
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -32,6 +34,25 @@ fn test_tool_with_save() -> Tool {
3234
.build()
3335
}
3436

37+
async fn spawn_malformed_chunked_server() -> String {
38+
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
39+
let addr = listener.local_addr().unwrap();
40+
41+
tokio::spawn(async move {
42+
if let Ok((mut stream, _)) = listener.accept().await {
43+
let mut buf = [0_u8; 1024];
44+
let _ = stream.read(&mut buf).await;
45+
let _ = stream
46+
.write_all(
47+
b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\nZZ\r\nboom\r\n0\r\n\r\n",
48+
)
49+
.await;
50+
}
51+
});
52+
53+
format!("http://{addr}/")
54+
}
55+
3556
#[tokio::test]
3657
async fn test_simple_get() {
3758
let mock_server = MockServer::start().await;
@@ -55,6 +76,28 @@ async fn test_simple_get() {
5576
assert_eq!(resp.format, Some("raw".to_string()));
5677
}
5778

79+
#[tokio::test]
80+
async fn test_malformed_chunked_body_returns_error() {
81+
let req = FetchRequest::new(spawn_malformed_chunked_server().await);
82+
let result = fetch_with_options(req, test_options()).await;
83+
84+
assert!(matches!(result, Err(FetchError::RequestError(_))));
85+
}
86+
87+
#[tokio::test]
88+
async fn test_save_to_file_malformed_chunked_body_does_not_create_empty_file() {
89+
let dir = tempfile::tempdir().unwrap();
90+
let saver = LocalFileSaver::new(Some(dir.path().to_path_buf()));
91+
let req =
92+
FetchRequest::new(spawn_malformed_chunked_server().await).save_to_file("malformed.txt");
93+
let result = test_tool_with_save()
94+
.execute_with_saver(req, Some(&saver))
95+
.await;
96+
97+
assert!(matches!(result, Err(FetchError::RequestError(_))));
98+
assert!(!dir.path().join("malformed.txt").exists());
99+
}
100+
58101
#[tokio::test]
59102
async fn test_head_request() {
60103
let mock_server = MockServer::start().await;

0 commit comments

Comments
 (0)