Skip to main content

max / makenotwork

storage: make multipart abort-on-failure structural Drive Storage A -> A+ (ultra-fuzz Run 7 --deep). Centralize the multipart abort into a single site. Once the multipart upload is created, upload_multipart delegates all fallible part/complete work to run_multipart_upload, which just returns Err; the wrapper aborts on any Err it returns. The four per-branch abort calls collapse to one, so a future failure path added inside the inner method aborts by construction and cannot forget to -- the Sto-S1 (5540d16) fix made structural rather than per-branch. Add a hermetic regression test for the pre-flight part-size guard (fires before the upload is created, so nothing is stranded and no network is hit). Gate: cargo clippy --all-targets + cargo test green for s3-storage; server crate compiles clean against it (public API unchanged, behavior preserved). No version bump, no deploy. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-25 00:46 UTC
Commit: 0cb8326515e54bcfa89ae6179b7791c471adffc5
Parent: 63e6a80
3 files changed, +95 insertions, -35 deletions
M shared/s3-storage/Cargo.lock +12
@@ -1956,10 +1956,22 @@ dependencies = [
1956 1956 "pin-project-lite",
1957 1957 "signal-hook-registry",
1958 1958 "socket2 0.6.3",
1959 + "tokio-macros",
1959 1960 "windows-sys 0.61.2",
1960 1961 ]
1961 1962
1962 1963 [[package]]
1964 + name = "tokio-macros"
1965 + version = "2.7.0"
1966 + source = "registry+https://github.com/rust-lang/crates.io-index"
1967 + checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496"
1968 + dependencies = [
1969 + "proc-macro2",
1970 + "quote",
1971 + "syn",
1972 + ]
1973 +
1974 + [[package]]
1963 1975 name = "tokio-rustls"
1964 1976 version = "0.24.1"
1965 1977 source = "registry+https://github.com/rust-lang/crates.io-index"
M shared/s3-storage/Cargo.toml +3
@@ -9,3 +9,6 @@ aws-config = { version = "1.8", features = ["behavior-version-latest"] }
9 9 bytes = "1"
10 10 tokio = { version = "1", features = ["fs", "io-util"] }
11 11 tracing = "0.1"
12 +
13 + [dev-dependencies]
14 + tokio = { version = "1", features = ["macros", "rt"] }
M shared/s3-storage/src/lib.rs +80 -35
@@ -417,7 +417,14 @@ impl S3Client {
417 417 /// Upload a file to S3 using multipart upload.
418 418 ///
419 419 /// Reads the file in `part_size` chunks (default 10 MB, minimum 5 MB) and
420 - /// uploads each as a part. Aborts the multipart upload on failure.
420 + /// uploads each as a part. Aborts the multipart upload on any failure.
421 + ///
422 + /// The abort lives in exactly ONE place: once the multipart upload has been
423 + /// created, all the fallible part/complete work runs in
424 + /// [`Self::run_multipart_upload`], and this wrapper aborts on any `Err` it
425 + /// returns. So a future failure path added inside the inner method aborts by
426 + /// construction — it cannot forget to (the Sto-S1 fix, made structural rather
427 + /// than per-branch).
421 428 pub async fn upload_multipart(
422 429 &self,
423 430 key: &str,
@@ -425,10 +432,9 @@ impl S3Client {
425 432 file_path: &std::path::Path,
426 433 part_size: Option<usize>,
427 434 ) -> Result<(), String> {
428 - use tokio::io::AsyncReadExt;
429 -
430 435 let part_size = part_size.unwrap_or(10 * 1024 * 1024); // 10 MB default
431 436 if part_size < 5 * 1024 * 1024 {
437 + // Pre-flight: nothing created yet, nothing to abort.
432 438 return Err("Multipart part size must be at least 5 MB".to_string());
433 439 }
434 440
@@ -447,6 +453,30 @@ impl S3Client {
447 453 .ok_or("S3 create multipart upload returned no upload_id")?
448 454 .to_string();
449 455
456 + // Single abort site: any failure past this point aborts exactly once.
457 + match self.run_multipart_upload(key, file_path, part_size, &upload_id).await {
458 + Ok(()) => Ok(()),
459 + Err(e) => {
460 + self.abort_multipart_upload(key, &upload_id).await;
461 + Err(e)
462 + }
463 + }
464 + }
465 +
466 + /// Read the file and upload every part, then complete the multipart upload.
467 + ///
468 + /// Returns `Err` (without aborting) on any failure; the caller
469 + /// [`Self::upload_multipart`] owns the single abort. Keep all abort handling
470 + /// out of here so the "abort on failure" contract can't be partially applied.
471 + async fn run_multipart_upload(
472 + &self,
473 + key: &str,
474 + file_path: &std::path::Path,
475 + part_size: usize,
476 + upload_id: &str,
477 + ) -> Result<(), String> {
478 + use tokio::io::AsyncReadExt;
479 +
450 480 let mut file = tokio::fs::File::open(file_path)
451 481 .await
452 482 .map_err(|e| format!("Failed to open file for multipart upload: {e}"))?;
@@ -462,10 +492,7 @@ impl S3Client {
462 492 match file.read(&mut buf[bytes_read..]).await {
463 493 Ok(0) => break,
464 494 Ok(n) => bytes_read += n,
465 - Err(e) => {
466 - self.abort_multipart_upload(key, &upload_id).await;
467 - return Err(format!("Failed to read file: {e}"));
468 - }
495 + Err(e) => return Err(format!("Failed to read file: {e}")),
469 496 }
470 497 }
471 498
@@ -489,7 +516,7 @@ impl S3Client {
489 516 .upload_part()
490 517 .bucket(&self.bucket)
491 518 .key(key)
492 - .upload_id(&upload_id)
519 + .upload_id(upload_id)
493 520 .part_number(part_number)
494 521 .body(body)
495 522 .send()
@@ -512,27 +539,20 @@ impl S3Client {
512 539 }
513 540 };
514 541
515 - match resp {
516 - Ok(resp) => {
517 - let etag = resp.e_tag().unwrap_or_default().to_string();
518 - completed_parts.push(
519 - CompletedPart::builder()
520 - .e_tag(etag)
521 - .part_number(part_number)
522 - .build(),
523 - );
524 - }
525 - Err(e) => {
526 - self.abort_multipart_upload(key, &upload_id).await;
527 - return Err(format!("S3 upload part {part_number} failed after retries: {e}"));
528 - }
529 - }
542 + let resp = resp
543 + .map_err(|e| format!("S3 upload part {part_number} failed after retries: {e}"))?;
544 + let etag = resp.e_tag().unwrap_or_default().to_string();
545 + completed_parts.push(
546 + CompletedPart::builder()
547 + .e_tag(etag)
548 + .part_number(part_number)
549 + .build(),
550 + );
530 551
531 552 part_number += 1;
532 553 }
533 554
534 555 if completed_parts.is_empty() {
535 - self.abort_multipart_upload(key, &upload_id).await;
536 556 return Err("No parts uploaded (empty file)".to_string());
537 557 }
538 558
@@ -540,23 +560,15 @@ impl S3Client {
540 560 .set_parts(Some(completed_parts))
541 561 .build();
542 562
543 - if let Err(e) = self
544 - .client
563 + self.client
545 564 .complete_multipart_upload()
546 565 .bucket(&self.bucket)
547 566 .key(key)
548 - .upload_id(&upload_id)
567 + .upload_id(upload_id)
549 568 .multipart_upload(completed)
550 569 .send()
551 570 .await
552 - {
553 - // Abort before returning, like every other failure branch (Sto-S1).
554 - // Without this, a failed completion leaves the multipart upload and all
555 - // its uploaded parts stranded server-side, accruing storage cost and
556 - // violating this function's "aborts on failure" contract.
557 - self.abort_multipart_upload(key, &upload_id).await;
558 - return Err(format!("S3 complete multipart upload failed: {e}"));
559 - }
571 + .map_err(|e| format!("S3 complete multipart upload failed: {e}"))?;
560 572
561 573 Ok(())
562 574 }
@@ -631,3 +643,36 @@ impl S3Client {
631 643 .map_err(|e| format!("{e}"))
632 644 }
633 645 }
646 +
647 + #[cfg(test)]
648 + mod tests {
649 + use super::*;
650 +
651 + fn test_client() -> S3Client {
652 + // `from_conf` is local — no network until a request is sent — so this
653 + // builds a usable client without reaching any endpoint.
654 + let s3_config = aws_sdk_s3::Config::builder()
655 + .behavior_version(BehaviorVersion::latest())
656 + .region(Region::new("test"))
657 + .endpoint_url("http://127.0.0.1:1")
658 + .credentials_provider(Credentials::new("ak", "sk", None, None, "test"))
659 + .force_path_style(true)
660 + .build();
661 + S3Client { client: Client::from_conf(s3_config), bucket: "test-bucket".to_string() }
662 + }
663 +
664 + #[tokio::test]
665 + async fn upload_multipart_rejects_undersized_part_before_any_request() {
666 + // The minimum-part-size guard is pre-flight: it must fire before the
667 + // multipart upload is created, so there is nothing to strand and no
668 + // network call (the dummy endpoint is unreachable — reaching it would
669 + // hang/error instead of returning this exact message).
670 + let client = test_client();
671 + let path = std::path::Path::new("/nonexistent");
672 + let err = client
673 + .upload_multipart("k", "application/octet-stream", path, Some(1024))
674 + .await
675 + .expect_err("undersized part size must be rejected");
676 + assert!(err.contains("at least 5 MB"), "unexpected error: {err}");
677 + }
678 + }