From 3cc0c9273704d880026554ccac959bb3f43589e8 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 26 May 2026 07:43:17 +0000 Subject: [PATCH] fix(cleanup): add retry logic to manifest reads and per-manifest error tolerance Three targeted fixes for S3 mid-stream body failures during cleanup: 1. read_manifest() now retries get_range() calls using the existing download_retry_count (default 3). Previously, manifest reads called object_store.inner.get_range() directly with no retry, while the CloudObjectReader path had retry via do_get_with_outer_retry(). Failed reads now log the full error details via log::warn/info. 2. process_manifests() no longer aborts on the first manifest read failure. Instead, it collects failures and retries them in up to 3 additional rounds. Only if manifests still fail after all retry rounds does the cleanup abort with a detailed error listing all failed paths and their errors. 3. ObjectStore.download_retry_count() is now public, allowing lance-table to access the configured retry count. Co-Authored-By: Jan van der Vegt --- rust/lance-io/src/object_store.rs | 5 ++ rust/lance-table/src/io/manifest.rs | 67 ++++++++++++++---- rust/lance/src/dataset/cleanup.rs | 105 ++++++++++++++++++++++++++-- 3 files changed, 160 insertions(+), 17 deletions(-) diff --git a/rust/lance-io/src/object_store.rs b/rust/lance-io/src/object_store.rs index 4327671b99b..3ed44274a1f 100644 --- a/rust/lance-io/src/object_store.rs +++ b/rust/lance-io/src/object_store.rs @@ -556,6 +556,11 @@ impl ObjectStore { .unwrap_or(self.io_parallelism) } + /// Number of times to retry a download that fails during body streaming. + pub fn download_retry_count(&self) -> usize { + self.download_retry_count + } + /// Get the IO tracker for this object store /// /// The IO tracker can be used to get statistics about read/write operations diff --git a/rust/lance-table/src/io/manifest.rs b/rust/lance-table/src/io/manifest.rs index 7df1d656263..57f739fd570 100644 --- a/rust/lance-table/src/io/manifest.rs +++ b/rust/lance-table/src/io/manifest.rs @@ -26,15 +26,59 @@ use crate::format::{DataStorageFormat, IndexMetadata, MAGIC, Manifest, Transacti use super::commit::ManifestLocation; +/// Wrapper around `object_store.inner.get_range()` that retries on transient +/// errors (e.g. mid-stream body download failures from S3). The upstream +/// `object_store` crate only retries on the initial HTTP request, not on +/// failures during body streaming. This mirrors the retry logic used by +/// `CloudObjectReader` in `lance-io`. +async fn get_range_with_retry( + object_store: &ObjectStore, + path: &Path, + range: Range, + retry_count: usize, +) -> Result { + let mut retries = retry_count; + loop { + match object_store.inner.get_range(path, range.clone()).await { + Ok(bytes) => return Ok(bytes), + Err(err) => { + if retries == 0 { + log::warn!( + "Failed to read manifest range {:?} from {} after {} attempts. \ + Error details: {:?}", + range, + path, + retry_count + 1, + err, + ); + return Err(err.into()); + } + log::info!( + "Retrying manifest read range {:?} from {} \ + (remaining retries: {}). Error: {:?}", + range, + path, + retries, + err, + ); + retries -= 1; + } + } + } +} + /// Read Manifest on URI. /// /// This only reads manifest files. It does not read data files. +/// Retries transient S3 errors (including mid-stream body failures) +/// up to `object_store.download_retry_count()` times per range request. #[instrument(level = "debug", skip(object_store))] pub async fn read_manifest( object_store: &ObjectStore, path: &Path, known_size: Option, ) -> Result { + let retry_count = object_store.download_retry_count(); let file_size = if let Some(known_size) = known_size { known_size } else { @@ -46,7 +90,7 @@ pub async fn read_manifest( start: initial_start, end: file_size, }; - let buf = object_store.inner.get_range(path, range).await?; + let buf = get_range_with_retry(object_store, path, range, retry_count).await?; // In case of corruption, the known_size might be wrong. We can retry without // the size to be more robust. @@ -75,18 +119,15 @@ pub async fn read_manifest( } else { // The prefetch only captured part of the manifest. We need to make an // additional range request to read the remainder. - let mut buf2: BytesMut = object_store - .inner - .get_range( - path, - Range { - start: manifest_pos as u64, - end: file_size - PREFETCH_SIZE, - }, - ) - .await? - .into_iter() - .collect(); + let remainder_range = Range { + start: manifest_pos as u64, + end: file_size - PREFETCH_SIZE, + }; + let mut buf2: BytesMut = + get_range_with_retry(object_store, path, remainder_range, retry_count) + .await? + .into_iter() + .collect(); buf2.extend_from_slice(&buf); buf2.freeze() }; diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index 464f735e2f6..45f23f877c3 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -194,19 +194,116 @@ impl<'a> CleanupTask<'a> { Ok(final_stats) } + /// Maximum number of retry rounds for manifests that fail during + /// `process_manifests`. Each round re-attempts all manifests that failed + /// in the previous round. + const MANIFEST_RETRY_ROUNDS: usize = 3; + #[instrument(level = "debug", skip_all)] async fn process_manifests( &'a self, tagged_versions: &HashSet, ) -> Result { let inspection = Mutex::new(CleanupInspection::default()); - self.dataset + let io_parallelism = self.dataset.object_store.io_parallelism(); + + // Collect all manifest locations first so we can retry failures. + let locations: Vec = self + .dataset .commit_handler .list_manifest_locations(&self.dataset.base, &self.dataset.object_store, false) - .try_for_each_concurrent(self.dataset.object_store.io_parallelism(), |location| { - self.process_manifest_file(location, &inspection, tagged_versions) - }) + .try_collect() .await?; + + let total = locations.len(); + info!( + "Processing {} manifest files (concurrency={})", + total, io_parallelism + ); + + let failed: Mutex> = Mutex::new(Vec::new()); + + // First pass: process all manifests, collecting failures instead of aborting. + stream::iter(locations) + .for_each_concurrent(io_parallelism, |location| { + let failed_ref = &failed; + let inspection_ref = &inspection; + async move { + if let Err(err) = self + .process_manifest_file(location.clone(), inspection_ref, tagged_versions) + .await + { + log::warn!("Failed to process manifest {}: {:?}", location.path, err,); + failed_ref + .lock() + .unwrap() + .push((location, format!("{:?}", err))); + } + } + }) + .await; + + // Retry rounds for failed manifests. + let mut pending_failures = std::mem::take(&mut *failed.lock().unwrap()); + for round in 1..=Self::MANIFEST_RETRY_ROUNDS { + if pending_failures.is_empty() { + break; + } + let retry_count = pending_failures.len(); + info!( + "Retry round {}/{}: re-processing {} failed manifest(s)", + round, + Self::MANIFEST_RETRY_ROUNDS, + retry_count, + ); + let round_failed: Mutex> = Mutex::new(Vec::new()); + + stream::iter(pending_failures.into_iter().map(|(loc, _)| loc)) + .for_each_concurrent(io_parallelism, |location| { + let round_failed_ref = &round_failed; + let inspection_ref = &inspection; + async move { + if let Err(err) = self + .process_manifest_file( + location.clone(), + inspection_ref, + tagged_versions, + ) + .await + { + log::warn!( + "Retry round {}: failed to process manifest {}: {:?}", + round, + location.path, + err, + ); + round_failed_ref + .lock() + .unwrap() + .push((location, format!("{:?}", err))); + } + } + }) + .await; + + pending_failures = std::mem::take(&mut *round_failed.lock().unwrap()); + } + + if !pending_failures.is_empty() { + let failed_paths: Vec = pending_failures + .iter() + .map(|(loc, err)| format!("{}: {}", loc.path, err)) + .collect(); + return Err(Error::io(format!( + "Failed to read {} manifest(s) after {} retry rounds: [{}]", + pending_failures.len(), + Self::MANIFEST_RETRY_ROUNDS, + failed_paths.join(", "), + ))); + } + + let succeeded = total; + info!("Successfully processed all {} manifests", succeeded); Ok(inspection.into_inner().unwrap()) }