diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs
index 6852d68..cdd912d 100644
--- a/src/backfill/worker/process.rs
+++ b/src/backfill/worker/process.rs
@@ -920,9 +920,59 @@ mod tests {
         url
     }
 
+    /// spawns a server answering getRepo with `chunks` as a chunked body, then dropping the
+    /// connection partway if `cut` is set
+    async fn spawn_streamed_get_repo(chunks: Vec<Bytes>, cut: bool) -> url::Url {
+        let app = Router::new().route(
+            "/xrpc/com.atproto.sync.getRepo",
+            get(move || {
+                use futures::StreamExt;
+                let parts = futures::stream::iter(chunks.clone().into_iter().map(Ok));
+                // late enough that the status and the first chunks are already out, so the
+                // client is reading the body when it breaks rather than still waiting on headers
+                let cut = futures::stream::iter(cut.then_some(())).then(|()| async {
+                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                    Err(std::io::Error::other("connection dropped"))
+                });
+                async move { axum::body::Body::from_stream(parts.chain(cut)) }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        url
+    }
+
+    /// a car of just its commit, or of a commit whose bytes don't match its cid
+    async fn commit_only_car(forged: bool) -> Bytes {
+        let commit = b"commit".to_vec();
+        let cid = jacquard_repo::mst::util::compute_cid(&commit).unwrap();
+        let mut bytes = Vec::new();
+        let mut writer =
+            iroh_car::CarWriter::new(iroh_car::CarHeader::new_v1(vec![cid]), &mut bytes);
+        let body: &[u8] = if forged { b"forged" } else { &commit };
+        writer.write(cid, body).await.unwrap();
+        writer.finish().await.unwrap();
+        Bytes::from(bytes)
+    }
+
+    fn tangled_only() -> FilteredCar {
+        let mut filter = crate::filter::FilterConfig::new(FilterMode::Filter);
+        filter.collections = vec![SmolStr::new("sh.tangled.repo")];
+        FilteredCar::new(Arc::new(filter), true).unwrap()
+    }
+
     async fn fetch(
         pds: &url::Url,
         max_body_bytes: usize,
+    ) -> Result<FullRepoOutcome, BackfillError> {
+        fetch_with(pds, max_body_bytes, None).await
+    }
+
+    async fn fetch_with(
+        pds: &url::Url,
+        max_body_bytes: usize,
+        filtered: Option<FilteredCar>,
     ) -> Result<FullRepoOutcome, BackfillError> {
         let address = std::net::SocketAddr::new("127.0.0.1".parse().unwrap(), pds.port().unwrap());
         let public_pds = url::Url::parse("http://pds.example/").unwrap();
@@ -944,11 +994,63 @@ mod tests {
             &throttle,
             &RateTier::trusted(),
             max_body_bytes,
-            None,
+            filtered,
         )
         .await
     }
 
+    #[tokio::test]
+    async fn streamed_get_repo_keeps_a_whole_car() {
+        let car = commit_only_car(false).await;
+        let pds =
+            spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
+                .await;
+        match fetch_with(&pds, 1024, Some(tangled_only())).await.unwrap() {
+            FullRepoOutcome::Filtered(repo) => assert_eq!(repo.stats.kept_blocks, 1),
+            other => panic!("expected Filtered, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn streamed_get_repo_maps_a_dropped_connection_to_transport() {
+        let car = commit_only_car(false).await;
+        let half = car.slice(..car.len() / 2);
+        let pds = spawn_streamed_get_repo(vec![half], true).await;
+        let err = fetch_with(&pds, 1024, Some(tangled_only()))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, BackfillError::Transport(_)), "{err}");
+    }
+
+    #[tokio::test]
+    async fn streamed_get_repo_rejects_a_forged_block_without_blaming_the_network() {
+        let pds = spawn_streamed_get_repo(vec![commit_only_car(true).await], false).await;
+        let err = fetch_with(&pds, 1024, Some(tangled_only()))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, BackfillError::Generic(_)), "{err}");
+    }
+
+    #[tokio::test]
+    async fn streamed_get_repo_stops_at_the_body_limit() {
+        let car = commit_only_car(false).await;
+        let max = car.len() - 1;
+        // a body that says up front how big it is, and one that only shows it as it arrives
+        let sized = spawn_get_repo(StatusCode::OK, car.to_vec()).await;
+        let chunked =
+            spawn_streamed_get_repo(car.chunks(7).map(Bytes::copy_from_slice).collect(), false)
+                .await;
+        for pds in [sized, chunked] {
+            let err = fetch_with(&pds, max, Some(tangled_only()))
+                .await
+                .unwrap_err();
+            assert!(
+                err.to_string().contains("exceeded max body size"),
+                "{pds}: {err}"
+            );
+        }
+    }
+
     #[cfg(feature = "indexer_stream")]
     #[tokio::test]
     async fn imported_identity_event_uses_the_reconciled_state() -> miette::Result<()> {