fix(storage): preserve multipart part identity across retries - #8174
Conversation
github-actions
Bot
added
A-encoding
![lance-gatekeeper[bot]](https://avatars.githubusercontent.com/in/4433141?s=60&v=4){kind=small}
There was a problem hiding this comment.
Gate recommendation: request changes. The native connector moves retries below part allocation correctly, but the change also removes transient multipart recovery from supported OpenDAL paths. Preserve the native fix and add an identity-safe retry at OpenDAL's writer boundary before removing the global fallback.
| } | ||
| } | ||
| Ok(Err(err)) => return Err(err.source.into()), | ||
| Ok(Err(err)) => return Err(err.into()), |
There was a problem hiding this comment.
The new immediate return removes the only transient multipart recovery for OpenDAL-backed stores. Native S3/Azure/GCS now retry inside their connectors, but use_opendal=true plus OpenDAL-only OSS/COS/TOS/GooseFS build bare Operator::from_iter values, and pinned OpenDAL installs no RetryLayer. A connection reset or RequestTimeout therefore aborts the whole upload instead of receiving the previous retry budget.
Keep trait-level re-entry disabled for native providers, but install identity-preserving retry at the OpenDAL writer boundary (e.g. RetryLayer, which retries the same writer and cloned buffer), with fault injection.
Reproducer
I ran cargo run --quiet against this head with a standalone store whose MultipartUpload::put_part returns Error::Generic(...ConnectionReset...) once and succeeds on its second invocation. The 5 MiB write drives ObjectWriter through multipart shutdown:
writer.write_all(&vec![0; 5 * 1024 * 1024]).await.unwrap();
let error = writer.shutdown().await.unwrap_err();
let observed = attempts.load(Ordering::SeqCst);
assert!(observed > 1, "transient multipart failure should be retried below provider part identity");Expected: the retry-capable provider path gets a second attempt. Observed:
observed attempts=1; error=Generic OpenDAL error: connection reset by peer
transient multipart failure should be retried below provider part identity
There was a problem hiding this comment.
No OpenDAL retry was added. The later maintainer review 4862887045 explicitly limits this bug to object_store and requires OpenDAL to remain unchanged; remote head c17039c restores every OpenDAL provider to its original path while retaining the native connector fix.
There was a problem hiding this comment.
Gate recommendation: request changes. The identity-safe writer retry now fixes the OpenDAL regression, but installing the 20-retry policy on the whole operator compounds ordinary transient failures with the outer AIMD retry loop. Keep retries at OpenDAL writer methods so multipart parts retain identity while reads, lists, deletes, and other operations continue to surface throttle feedback to AIMD.
| .with_max_delay(Duration::from_secs(8)) | ||
| .with_max_times(max_retries()) | ||
| .with_jitter(); | ||
| OpendalStore::new(operator.layer(retry_layer)) |
There was a problem hiding this comment.
RetryLayer wraps the entire Operator, so the multipart recovery budget also applies to stat, reads, lists, deletes, copies, and renames before the outer AIMD layer sees an error. With the defaults, a temporary throttle response receives 21 OpenDAL attempts; AIMD then repeats that exhausted logical operation three more times, delaying shared rate reduction and producing 84 provider requests from one call. The exponential 2/4/8-second schedule plus additive jitter can also turn a single ordinary operation into a many-minute wait. Use an OpenDAL Layer/Service wrapper that passes access operations through unchanged and applies stateful retry only to the returned oio::Writer methods.
Reproducer
At this head I changed the existing synthetic service stat method to increment attempts and always return Error::new(ErrorKind::Unexpected, "429 Too Many Requests").set_temporary(), then added:
let store = Arc::new(store_with_retry(Operator::new(builder).unwrap()));
let store = AimdThrottledStore::new(store, AimdThrottleConfig::default()).unwrap();
let _ = store.head(&Path::from("object")).await.unwrap_err();
assert_eq!(attempts.load(Ordering::SeqCst), 1,
"multipart-writer retry must not retry or hide ordinary HEAD failures");I ran cargo test -p lance-io --all-features object_store::opendal_retry::tests::reproduce_retry_layer_scope_and_aimd_compounding -- --exact --nocapture. The assertion failed with left: 84, right: 1.
There was a problem hiding this comment.
❌ Gate recommendation: request changes.
The writer-only boundary resolves the previous AIMD amplification, but the generic close retry is unsafe for destructive writer state machines and can convert failed nonempty writes into successful empty objects. Limit recovery to replay-safe write operations, or make retry eligibility provider- and operation-aware; close and abort should remain one-shot unless the provider preserves staged state after an error.
| } | ||
|
|
||
| async fn close(&mut self) -> Result<Metadata> { | ||
| retry_writer_operation(&mut self.inner, |writer| Box::pin(writer.close())).await |
There was a problem hiding this comment.
Retrying this close can turn nonempty GooseFS output into a successfully published empty object. The pinned GoosefsWriter::close removes self.writer before awaiting the SDK close; on error it takes and deletes tmp_path before returning. TransportError is marked temporary, so this wrapper calls close again on empty state, which enters the GooseFS zero-write branch and publishes an empty file.
Keep close one-shot, matching the removed ObjectWriter behavior, and do the same for abort unless a provider guarantees replayability. Alternatively, enable retries only for provider/method pairs whose state remains resumable; the part-recovery failure addressed here only requires retrying write.
Reproducer
A live GooseFS transport fault requires a cluster and private SDK error construction, so I ran the closest bounded state-machine test against this head. I added an oio::Write whose close mirrors GooseFS by consuming buffered state before returning one temporary error:
async fn close(&mut self) -> Result<Metadata> {
let buffered_size = std::mem::take(&mut self.buffered_size);
if self.close_attempts.fetch_add(1, Ordering::SeqCst) == 0 {
Err(Error::new(ErrorKind::Unexpected, "transport closed").set_temporary())
} else {
self.published_size.store(buffered_size, Ordering::SeqCst);
Ok(Metadata::new(EntryMode::FILE))
}
}
let mut writer = RetryWriter { inner: destructive_writer };
writer.write(Buffer::from("payload")).await.unwrap();
writer.close().await.unwrap();
assert_eq!(published_size.load(Ordering::SeqCst), 7);I ran cargo test -p lance-io --all-features object_store::opendal_retry::tests::reproduce_destructive_close_retry_publishes_empty_object -- --exact --nocapture. The assertion failed with left: 0, right: 7 after two close attempts.
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
There was a problem hiding this comment.
❌ Gate recommendation: request changes.
The destructive GooseFS finalization replay is fixed, but retrying only write does not recover the common OpenDAL multipart failures this PR is intended to handle: OpenDAL queues part work and surfaces temporary failures while close drains it.
Make the retry boundary provider- and state-machine-aware: HTTP multipart/block writers need recovery at a stable part/task boundary, while GooseFS-style destructive finalization must remain one-shot.
| async fn close(&mut self) -> Result<Metadata> { | ||
| // Finalization can consume staged state before returning an error, so it is not | ||
| // replay-safe. | ||
| self.inner.close().await |
There was a problem hiding this comment.
This one-shot close drops transient recovery for HTTP-backed OpenDAL multipart writers. In pinned OpenDAL 0.58.1, MultipartWriter::write caches or enqueues part work and commonly returns Ok(()); MultipartWriter::close submits and drains those tasks. If a queued part returns a temporary connection error, ConcurrentTasks::next preserves and requeues the same part input but returns the error so the caller can re-enter close. This wrapper instead returns that first error, so common one- or few-part OSS/COS/TOS uploads—and OpenDAL S3/Azure/GCS when AIMD does not classify the error as throttling—fail without using the configured retry budget.
Make finalization policy aware of the underlying writer: retry resumable OpenDAL multipart/block task draining, or retry at its stable part-request boundary, while keeping destructive GooseFS close one-shot.
Reproducer
I ran this regression against the pinned oio::MultipartWriter at this head. Its first part attempt returns one temporary connection error; the writer retains that task for a retry, but this wrapper does not re-enter close:
#[derive(Debug)]
struct DeferredPartWriter {
attempts: Arc<AtomicUsize>,
}
impl oio::MultipartWrite for DeferredPartWriter {
async fn write_once(&self, _: u64, _: Buffer) -> Result<Metadata> {
Ok(Metadata::new(EntryMode::FILE))
}
async fn initiate_part(&self) -> Result<String> {
Ok("upload".to_string())
}
async fn write_part(
&self, _: &str, part_number: usize, _: u64, _: Buffer,
) -> Result<oio::MultipartPart> {
if self.attempts.fetch_add(1, Ordering::SeqCst) == 0 {
return Err(Error::new(
ErrorKind::Unexpected, "connection reset by peer",
).set_temporary());
}
Ok(oio::MultipartPart {
part_number, etag: format!("part-{part_number}"),
checksum: None, size: None,
})
}
async fn complete_part(
&self, _: &str, _: &[oio::MultipartPart],
) -> Result<Metadata> {
Ok(Metadata::new(EntryMode::FILE))
}
async fn abort_part(&self, _: &str) -> Result<()> { Ok(()) }
}
#[tokio::test(start_paused = true)]
async fn reproduce_deferred_part_failure_escapes_write_retry() {
let attempts = Arc::new(AtomicUsize::new(0));
let inner = oio::MultipartWriter::new(
opendal::Executor::default(),
DeferredPartWriter { attempts: Arc::clone(&attempts) },
8,
);
let mut writer = RetryWriter { inner };
writer.write(Buffer::from("part-0")).await.unwrap();
writer.write(Buffer::from("part-1")).await.unwrap();
writer.close().await.unwrap();
assert!(attempts.load(Ordering::SeqCst) > 2);
}I ran cargo test -p lance-io --all-features object_store::opendal_retry::tests::reproduce_deferred_part_failure_escapes_write_retry -- --exact --nocapture. Expected: finalization succeeds after retrying the preserved part task. Observed: close().unwrap() failed on the first attempt with temporary connection reset by peer.
There was a problem hiding this comment.
❌ Gate recommendation: request changes.
The provider-aware close policy resolves the prior finalization failures, but GooseFS still inherits unconditional whole-buffer write replay. Because its writer accepts buffers incrementally, a temporary mid-buffer failure can duplicate already-written bytes.
Classify write replay independently: bypass generic write retry for GooseFS, or retry only below a provider boundary that can prove the failed bytes were not accepted.
|
|
||
| impl<W: oio::Write> oio::Write for RetryWriter<W> { | ||
| async fn write(&mut self, body: Buffer) -> Result<()> { | ||
| retry_replay_safe_operation(&mut self.inner, "write", move |writer| { |
There was a problem hiding this comment.
This retries writer.write by replaying the entire Buffer, but the GooseFS writer can consume a prefix before a later chunk fails. The same writer has then advanced, so replaying the original buffer duplicates the accepted prefix and silently corrupts object data. FinalizationRetry::OneShot only gates close; GooseFS still uses this write path.
Classify write replay independently: disable it for GooseFS, or retry below a provider boundary that can prove the failed bytes were not accepted.
Reproducer
At this head I added a synthetic incremental writer that accepted "prefix", returned one temporary error, and then accepted the retried buffer:
writer.write(Buffer::from("prefixsuffix")).await.unwrap();
assert_eq!(written, b"prefixsuffix");I ran:
cargo test -p lance-io --all-features object_store::opendal_retry::tests::reproduce_partial_write_replay_duplicates_prefix -- --exact --nocapture
Expected: prefixsuffix. Observed: prefixprefixsuffix.
There was a problem hiding this comment.
Addressed in c17039c. Removed the generic OpenDAL writer retry layer, so GooseFS no longer replays partially accepted buffers; every OpenDAL provider now uses its original OpendalStore path.
Xuanwo
left a comment
Xuanwo
left a comment
There was a problem hiding this comment.
This bug is object_store only, please don't touch opendal.
|
Addressed in c17039c. Removed the OpenDAL retry module and restored every OpenDAL provider and writer path. The remaining PR diff is limited to native |
There was a problem hiding this comment.
✅ Gate recommendation: approve.
The native connector now retries below multipart part allocation, preserving provider-assigned identities while feeding each physical attempt into the shared write AIMD controller. Removing generic put_part resubmission also avoids unsafe replay for OpenDAL and custom stores.
1 participant
Summary
object_storeHTTP connector so every attempt retains the provider-assigned part identityRoot cause
The AIMD layer retried a failed upload by invoking
MultipartUpload::put_partagain. Native cloud stores allocate a new part number when that method is called, so the retry skipped the failed part and completion reportedMissing part.Native S3, Azure, and GCS now retry the same HTTP request after part allocation. The outer multipart wrapper recognizes those native stores and does not retry the logical part again. OpenDAL stores retain their existing behavior and are outside this repair.
Validation
cargo fmt --allandcargo fmt --all -- --checkcargo clippy --all --tests --benches -- -D warningscargo clippy -p lance-io --all-features --tests --benches -- -D warningscargo test -p lance-io --all-features test_multipart_http_retry_reuses_part_number -- --nocapture(1 passed)cargo test -p lance-io --lib --all-features -- --test-threads=1 --skip uring::tests(251 passed)git diff --checkFixes #7956