From 4f5697fdd9ff99e2d9b4c00648b6361d166a2592 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 26 Sep 2026 17:31:26 -0500 Subject: [PATCH] clawhdf5-remote: readers waiting on a failed fetch get its error A reader that waited for another reader's fetch of a block got "the fetch of this block failed" when that fetch failed, not why: a file replaced on the server while open was reported as FileChanged to one thread and as an anonymous failure to the others. The fetch's error is now handed to every reader waiting on it. Regression test: four threads read the same block from a slow backend whose fetches fail with a "changed while open" error; each gets that error (it failed for the waiters before this change). Also fixes the ignore-Range test, broken by the previous commit: the test server now counts a body before sending it, so "the refused body was not read" is checked as "refused at the first response". Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/clawhdf5-remote/src/cache.rs | 73 ++++++++++++++++++++++------ crates/clawhdf5-remote/tests/http.rs | 6 +-- 2 files changed, 58 insertions(+), 21 deletions(-) diff --git a/crates/clawhdf5-remote/src/cache.rs b/crates/clawhdf5-remote/src/cache.rs index 233938c..a434814 100644 --- a/crates/clawhdf5-remote/src/cache.rs +++ b/crates/clawhdf5-remote/src/cache.rs @@ -98,7 +98,7 @@ type Block = Arc<[u8]>; /// A fetch in progress: the readers waiting for a block wait on this. struct Flight { - result: Mutex>>, + result: Mutex>>, done: Condvar, } @@ -110,7 +110,7 @@ impl Flight { }) } - fn finish(&self, r: Result) { + fn finish(&self, r: Result) { let mut slot = lock(&self.result); if slot.is_none() { *slot = Some(r); @@ -118,7 +118,7 @@ impl Flight { self.done.notify_all(); } - fn wait(&self) -> Result { + fn wait(&self) -> Result { let mut slot = lock(&self.result); loop { if let Some(r) = slot.as_ref() { @@ -173,10 +173,20 @@ pub struct BlockCache { } /// Fails every flight a fetch claimed and did not complete (an error or a -/// panic in the backend), so no reader waits forever. +/// panic in the backend), so no reader waits forever. The waiting readers +/// get the fetch's error. struct FlightGuard<'a, S> { cache: &'a BlockCache, flights: Vec<(u64, Arc)>, + error: Option, +} + +impl FlightGuard<'_, S> { + /// Record `e` for the readers waiting on this fetch; returns it. + fn fail(&mut self, e: FormatError) -> FormatError { + self.error = Some(e.clone()); + e + } } impl Drop for FlightGuard<'_, S> { @@ -191,8 +201,12 @@ impl Drop for FlightGuard<'_, S> { } } drop(st); + let error = self + .error + .take() + .unwrap_or_else(|| FormatError::Storage("the fetch of this block failed".into())); for (_, f) in &self.flights { - f.finish(Err("the fetch of this block failed".into())); + f.finish(Err(error.clone())); } } } @@ -348,6 +362,7 @@ impl BlockCache { let mut guard = FlightGuard { cache: self, flights: Vec::new(), + error: None, }; { let mut st = lock(&self.state); @@ -410,13 +425,16 @@ impl BlockCache { self.counters .requests .fetch_add(runs.len() as u64, Ordering::Relaxed); - let fetched = self.inner.read_ranges(&runs)?; + let fetched = match self.inner.read_ranges(&runs) { + Ok(f) => f, + Err(e) => return Err(guard.fail(e)), + }; if fetched.len() != runs.len() { - return Err(FormatError::Storage(format!( + return Err(guard.fail(FormatError::Storage(format!( "backend returned {} ranges for {} requested", fetched.len(), runs.len() - ))); + )))); } let mut got: Vec<(u64, Block)> = Vec::with_capacity(guard.flights.len()); for (run, bytes) in runs.iter().zip(&fetched) { @@ -424,12 +442,12 @@ impl BlockCache { .bytes_fetched .fetch_add(bytes.len() as u64, Ordering::Relaxed); if bytes.len() as u64 != run.end - run.start { - return Err(FormatError::Storage(format!( + return Err(guard.fail(FormatError::Storage(format!( "short read from the backend: {} of {} bytes at offset {}", bytes.len(), run.end - run.start, run.start - ))); + )))); } let bs = self.config.block_size; let mut i = run.start / bs; @@ -476,12 +494,7 @@ impl BlockCache { } for (i, f) in waits { - match f.wait() { - Ok(data) => { - have.insert(i, data); - } - Err(e) => return Err(FormatError::Storage(e)), - } + have.insert(i, f.wait()?); } Ok(have) } @@ -786,6 +799,34 @@ mod tests { } } + /// A slow backend whose fetches all fail with a telling error. + struct SlowFailing; + + impl Storage for SlowFailing { + fn read_at(&self, _: u64, _: usize) -> Result, FormatError> { + std::thread::sleep(std::time::Duration::from_millis(50)); + Err(FormatError::Storage("the file changed while open".into())) + } + fn len(&self) -> u64 { + 1 << 20 + } + } + + #[test] + fn readers_waiting_on_a_failed_fetch_get_its_error() { + let c = BlockCache::new(SlowFailing, small()); + std::thread::scope(|s| { + let hs: Vec<_> = (0..4) + .map(|_| s.spawn(|| c.read_at(0, 10).map(|b| b.len()))) + .collect(); + for h in hs { + let e = h.join().unwrap().unwrap_err().to_string(); + assert!(e.contains("changed while open"), "{e}"); + } + }); + assert!(c.stats().waits > 0); + } + #[test] fn concurrent_readers_never_fetch_a_block_twice() { let d = data(64 * 1024); diff --git a/crates/clawhdf5-remote/tests/http.rs b/crates/clawhdf5-remote/tests/http.rs index b0a0e34..5a68072 100644 --- a/crates/clawhdf5-remote/tests/http.rs +++ b/crates/clawhdf5-remote/tests/http.rs @@ -301,11 +301,7 @@ fn a_server_that_ignores_range_is_refused_or_downloaded_when_allowed() { matches!(err, Error::Remote(RemoteError::RangeNotSupported(_))), "{err}" ); - assert!( - server.bytes() < bytes.len() as u64, - "refusing must not download the file ({} bytes read)", - server.bytes() - ); + assert_eq!(server.requests(), 1, "refused at the first response"); let mut opts = quick(); opts.http.allow_full_download = true; let storage = storage_for_url(&url, &opts).unwrap();